Re: Making multiple instances of a RMI method in (from?) one server implementation

On Wednesday 06 June 2007 06:13, Lenny Wintfeld wrote:
> I have a partial design which I've implemented in skeletal  form just to
> get a better understanding of RMI and remote synchronization. I have a
> single  "distributor" program and a set (actually 2 for now) of  data
> "processor" programs. The "distributor" has a threadsafe incoming queue
> which  holds the names of  data "processor" programs that are available
> to process a data set. The names are submitted to the queue by the
> "processor" programs when they become idle. I pull names out of that
> queue and call the instance of the data "processor" program
> corresponding to that name using  RMI. Since the one "distributor" and
> the multiple "processor" programs access services from one another using
> RMI they act as both server and client to one another. At this point I
> have this bare skeleton of  a  subsystem (without doing any actual
> processing of data) working ok.

Your problem is typical for what is known as JavaSpaces, which allows a much
simpler abstraction for you (once you understand it, and knows how to set it
up). JavaSpaces is a kind of network-shared memory, where you can
put/get/read named objects.
JavaSpaces is based on Jini[1], which will handle all the network tricky bits,
such as discovery, leasing and remote events.

> The two inter-related design problems I'd like advice on have to do with
> devising a method for the data "processor" instances to return their
> result sets to the "distributor" once they're done.

RMI method calls are not much different from standard Java calls. If you have
handed out multiple references to the "distributor" singleton, then all of
those clients can call the methods on that instance at the same time. The
call to the "distributor" is not made until all the data is transferred from
the client, so the "distributor" method will always execute full or not at
all, so it is not like it will interrupt half way through the method call if
the connection is closed (for instance).

I think you real problem is to deal with the situation that a "processor" dies
before completing the task, and that has to be detected as well as the task
has to be handed to someone else. Typical JavaSpaces problem.

Problem 1: Is not a problem.

> Problem 2: The results set produced by the "processor" may be quite
> large.  Large enough that the results can't be sent as the RMI method
> return value from the call  to the "processor". The vm runs out of
> memory (at least in Eclipse it does) when I try and return a a test byte
> array of 10MB from the "processor" method. Instead, I'm thinking that
> I'd  like to send the results over to the "distributor" piecemeal, by
> defining a "resultReciever" RMI method in the "distributor" that the
> "processor" could call as many times as necessary to move its whole
> result set.  The method signature would look something like
>
> public void resultReceiver(int blockNr, int blockTotal, int byteCnt,
> byte[] result);
>
> Where blockNr is the sequence number of the current block being sent,
> blockTotal is the number of the last block that will be sent, byteCnt is
> the size of the current block being sent and result is the data block
> itself. To make sure that no blocks are missing I'd keep a lastBlockNr
> in each instance of the object that implements the thread containing the
> resultReceiver method at the "distributor". Does this sound like a
> reasonable approach?

Nah. First off; byteCnt is not needed. byte[] result already knows that.
Secondly, if you are dealing with 100s MB of data, the yes, I suggest that you
don't put that in byte[]. Instead, let the "processor" stream the data back
via an ordinary socket.

byte[] result = ...;
try
{
    Socket s = new Socket( distributorIP, RESULTPORT );
    s.connect();
    BufferedOutputStream out =
        new BufferedOutputStream( s.getOutputStream() ) );
    out.writeLong( batchId ); // identifying the batch
    out.writeInt( result.length ); // size of data
    out.write( result );
    out.flush();
} finally
{
    out.close();
}

The "distributor" would wait around for connections

ServerSocket listen = new ServerSocket( RESULTPORT );
while( running )
{
    Socket s = listen.accept();
    MyThread t = threadPool.get();
    t.process( s );
}

public class MyThread
    implements Runnable
{
    private Socket socket;

    public void run()
    {
        try
        {
            while( true )
            {
                synchronized( this )
                {
                    wait(); // Wait for job.
                    processResult();
                }
            }
        } catch( Exception e )
        {
           // Do something.
        }
    }

   void process( Socket s )
   {
       synchronized( this )
       {
           socket = s;
           notify();
       }
   }

   private void processResult()
       throws IOException, ClassNotFoundException,...
   {

       BufferedInputStream in = new BufferedInputStream(
           s.getInputStream()
       );
       long batchId = in.readLong();
       int size = in.readInt();
       byte[] result = new byte[size];
       in.read( result );
   }
}

Very rough sketch, but you should get the idea.
It still doesn't solve time-outs and resilience against "processor" failures.


[1] https://jini.dev.java.net/
    http://incubator.apache.org/river


Cheers
--
Niclas Hedhman, Software Developer

I  live here; http://tinyurl.com/2qq9er
I  work here; http://tinyurl.com/2ymelc
I relax here; http://tinyurl.com/2cgsug

===========================================================================
To unsubscribe, send email to [email protected] and include in the body
of the message "signoff RMI-USERS".  For general help, send email to
[email protected] and include in the body of the message "help".

For a list of frequently asked RMI questions please refer to:
http://java.sun.com/j2se/1.3/docs/guide/rmi/faq.html

To view past RMI-USERS postings, please see:
http://archives.java.sun.com/archives/rmi-users.html