As a pluggable daemon, mcp needs a flexible command syntax to permit both control of the plugins and passthrough of commands to the plugins for scripting.

First, we make a usage function to advise the user:

        def usage(argv,stdin,stdout,stderr)
                stderr.puts "Usage: mcp {command}
Where {command} is one of:

        plugin stop {plugin name}           - Stop a plugin thread
        plugin start {plugin name}          - Start a plugin thread
        plugin load {plugin name}           - Load a named plugin
        plugin unload {plugin name}         - Unload a named plugin
        plugin tell {plugin name} {command} - Tell a plugin a command
        thread list                         - List currently running threads
        exit                                - Kill mcpd
"
                1
        end

You probably want to use a "here document" for that multi-line print, but I'm having problems getting it to render in bluecloth (markdown) at the moment.

Now for the real fun. All of the commands are passed to the command() method. This is where we handle each of the above:

        def command(argv,stdin,stdout,stderr)
                @command=argv.join(' ')
                begin
                        log("mcp #{@command}")
                        case @command
                                when /^quit$/i, /^exit$/i
                                        # Need more exit handling here!
                                        exit
                                when /^plugin list$/i
                                        @plugins.each_key { |plugin| stdout.puts "#{plugin}\n" }
                                when /^thread list$/i
                                        stdout.puts Thread.list.map { |t| "#{t.to_s} #{t['name']}\n" }
                                when /^plugin tell (\S+) (.*)$/i
                                        log("Telling #{$1} to #{$2}")
                                        @plugins[$1].command($2,stdin,stdout,stderr)
                                when /^plugin start (.*)$/i
                                        @plugins[$1].start()
                                when /^plugin stop (.*)$/i
                                        @plugins[$1].stop()
                                when /^plugin load (.*)$/i
                                        plugin_load($1)
                                when /^plugin unload (.*)$/
                                        plugin_unload($1)
                                else
                                        usage(argv,stdin,stdout,stderr);
                        end
                rescue => detail
                        stderr.puts detail.message + "\n"
                        stderr.puts detail.backtrace.join("\n") + "\n"
                        1
                end
        end

Simple, eh? Now plugins are controllable from the command line.

Not bad for ~100 lines of ruby so far.

The next step is setting the thread['name'] properties for the "thread list" command. I'll cover that in the next post.

The next step in MCP development is getting the plugins to load/unload dynamically.

Each plugin will be a .rb file in the "plugins.d/" directory. Each plugin will consist of a class named like the file, containing 5 basic methods:

• initialize() Called whenever a plugin is loaded.
• start() Called to start a plugin.
• command() Called by mcpd to relay commands from mcp.
• stop() Called to stop a plugin (stop threads)
• unload() Called to unload a plugin cleanly

The loaded plugin must also have an attribute "mtime", which will be used by mcp to track the mtime on the loaded file to make sure it matches that of the .rb file in the "plugins.d/" directory during the periodic dynamic plugin scan.

Here is an example plugin called Tron, as it would appear in "plugins.d/Tron.rb":

class Tron
        attr_accessor :running, :mtime, :thread, :daemon

        def command(arg,stdin,stdout,stderr)
                @daemon.log("Tron received command #{arg}")
                "Greetings user. Nothing to report.\n"
        end

        def unload
                @running=false
        end

        def stop
                @running=false
                @thread.join
                "Tron thread stopped.\n"
        end

        def start
                if not @running
                        @running=true
                        @thread=Thread.new do
                                @daemon.log("Tron thread started (#{@mtime})")
                                Thread.current['name']="Tron thread"
                                while @running do
                                        sleep 10
                                end
                                @daemon.log("Tron thread ending (#{@mtime})")
                        end
                        "Tron thread started\n"
                else
                        "Tron thread already running\n"
                end
        end

        def initialize(daemon)
                @daemon=daemon
                start()
        end

end

Now to add the dynamic plugin functionality to mcpd.

The first step is to define some constants and create an attribute to enumerate all loaded plugins.

#
# Constants
#
PLUGINS_DIR = 'plugins.d'

class MCPDaemon
        attr_accessor :plugins

Now we will add a method for loading plugins:

        def plugin_load(name)
                if not @plugins[name]
                        @file="#{PLUGINS_DIR}/#{name}.rb"
                        begin
                                load "#{@file}"
                                eval "@plugin=#{name}.new(self)"
                                @plugin.mtime=File.stat(@file).mtime
                                @plugins[name]=@plugin
                                log("Loaded plugin: #{name} (#{@plugin.mtime})")
                        rescue => detail
                                log("Error loading plugin: #{name}: #{detail.message}")
                        end
                end
                @plugins.has_key?(name)
        end

This will load a plugin if it is not already loaded, and will return the boolean true if the plugin is successfully loaded.

The plugin_unload method likewise unloads plugins on demand by calling the plugin's unload() method and deleting it from the @plugins hash.

        def plugin_unload(name)
                if @plugins[name]
                        begin
                                @plugins[name].unload()
                                log("Unloaded plugin: #{name} (#{@plugins[name].mtime})")
                                @plugins.delete(name)
                        rescue => detail
                                log("Error unloading plugin: #{name}: #{detail.message}")
                        end
                        not @plugins.has_key?(name)
                end
        end

The most complex method is plugins_scan(). It is responsible for checking the mtime of each plugin's source file against the loaded mtime of each plugin.

        def plugins_scan()
                begin
                        @plugin_files={}

                        Dir["#{PLUGINS_DIR}/*"].each do |file|
                                name=File.basename(file).sub!(/\.rb\Z/,'')
                                @plugin_files[name]=file
                        end

                        @plugin_files.each do |name,file|
                                if @plugins && @plugins.has_key?(name)
                                        if File.stat(file).mtime != @plugins[name].mtime
                                                plugin_unload(name)
                                                plugin_load(name)
                                        end
                                else
                                        plugin_load(name)
                                end
                        end

                        @plugins.each do |name,plugin|
                                if not @plugin_files.has_key?(name)
                                        plugin_unload(name)
                                end
                        end
                rescue => detail
                        log(detail.backtrace.join("\n"))
                end
        end

The final bit of code is added to MCPDaemon's initialize() method to initialize the @plugins hash and start the plugin scanner thread:

                @plugins={}

                # Plugin handler thread
                Thread.new do
                        while true do
                                plugins_scan()
                                sleep 1
                        end             
                end     

Now, go ahead and start mcpd, you should some output like the following:

$ ./mcpd
mcpd: Tron thread started ()
mcpd: Loaded plugin: Tron (Thu Nov 10 00:54:54 EST 2005)
mcpd: Daemon started

If you touch the plugin, it will automagically reload:

mcpd: Unloaded plugin: Tron (Thu Nov 10 00:54:54 EST 2005)
mcpd: Tron thread started ()
mcpd: Loaded plugin: Tron (Thu Nov 10 00:56:22 EST 2005)
mcpd: Tron thread ending (Thu Nov 10 00:54:54 EST 2005)

Likewise, if you remove the plugin, it is unloaded:

mcpd: Unloaded plugin: Tron (Thu Nov 10 00:56:22 EST 2005)
mcpd: Tron thread ending (Thu Nov 10 00:56:22 EST 2005)

Fun! Next step: command() syntax for mcp

Distributed Ruby (DRb)'s closest equivalent is probably Java's Remote Method Invocation (RMI). Written in only 200 lines of native Ruby, DRb is an incredibly easy way of communicating between processes in disparate systems.

For the Master Control Program (MCP) project, I needed a way for the command line "mcp" userspace tool to talk to the running "mcpd" daemon. Rather than make this a true heavyweight client/server setup, I plan on making the "mcp" client little more than a stub that talks to the "heavy" server. Using this method, far less code is actually required to provide the same function.

To that end, the "mcp" client script initially looked like this:

#!/usr/bin/env ruby

require 'drb'
require 'drb/unix'

begin
        DRb.start_service
        mcpd = DRbObject.new nil, "drbunix:/tmp/mcpd.sock"
        print mcpd.command(ARGV)
rescue
        print "ERROR: unable to communicate with mcpd\n"
end

and the server side was little more than this:

#!/usr/bin/env ruby

require 'syslog'
require 'drb'
require 'drb/unix'

class MCPDaemon
        def log(message)
                @syslog.info(message)
        end  

        def command(argv)
                @command=argv.join(' ')
                log("Client said to run #{@command}")
                "You said to run #{@command}"
        end

        def initialize
                @syslog=Syslog.open( File.basename(__FILE__),
                                                   Syslog::LOG_PERROR |
                                                   Syslog::LOG_NDELAY )
                # More initialization here.
                DRb.start_service "drbunix:/tmp/mcpd.sock", self
                log("Daemon started")
                DRb.thread.join
        end
end

MCPDaemon.new

If you spawn "mcpd" from one window, you can run "mcp" from another window and watch mcpd log the commands you tell it to run.

This doesn't fulfil the entire problem, however. In its current form, calls to the running mcpd's "command" method are marshalled. This means that ARGV is copied when being passed to command() remotely, and the return string is copied back as well.

Ideally, we want to be able to stream stdin/stdout/stderr from the mcp client to the running mcpd daemon by handing over object references. The trick to this is to mark those objects as "DRbUndumped" and pass them along on the function call. This will proxy those objects instead of marshalling them, treating them as if they were local:

#!/usr/bin/env ruby

require 'drb'
require 'drb/unix'

begin
        DRb.start_service
        mcpd = DRbObject.new nil, "drbunix:/tmp/mcpd.sock"
        $stdin.extend DRbUndumped
        $stdout.extend DRbUndumped
        $stderr.extend DRbUndumped
        return mcpd.command(ARGV, $stdin, $stdout, $stderr)
rescue
        print "ERROR: unable to communicate with mcpd\n"
end

Now the running mcpd can be modified to use these new proxied objects directly:

#!/usr/bin/env ruby

require 'syslog'
require 'drb'
require 'drb/unix'

class MCPD
        def log(message)
                @syslog.info(message)
        end  

        def command(argv,stdin,stdout,stdout)
                @command=argv.join(' ')
                begin
                        log("Client said to run #{@command}")
                        stdout.puts "You said to run #{@command}\n"
                        0
                rescue => detail
                        stderr.puts detail.message + "\n"
                        stderr.puts detail.backtrace.join("\n") + "\n"
                        1
                end
        end

        def initialize
                @syslog=Syslog.open( File.basename(__FILE__),
                                                   Syslog::LOG_PERROR |
                                                   Syslog::LOG_NDELAY )
                # More initialization here.
                DRb.start_service "drbunix:/tmp/mcpd.sock", self
                log("Daemon started")
                DRb.thread.join
        end
end

MCPDaemon.new

Whenever those proxied objects are referenced in the command() method, the client's stdin/stdout/stderr are actually used.

Neat, huh? :)

Next step: plugins.