How do you execute new code in an existing Thread?

[Ken Blair]

In other words, how would you execute methods in an existing thread when you won't know until runtime which method(s) need executing or n which order and it will be determined by a different thread?

What I'm working on is an XML-RPC server using Apache's implementation. I'm working with COBOL files and in order to open a new connection to a file it must be done in a new Thread. The fun part comes when I have to keep that Thread alive and wait for further requests to be made on that file. I don't know, of course, until later when the client makes the request what method(s) are going to be called or when. Once the request comes in, however, I must execute the code from the same thread that it was opened in. It looks something like this:

1. A request comes into the server from Client A.
2. The request is spawned off in Thread A by the ZML-RPC server and it executes a handler.
3. The handler's execution creates Thread 1, which opens a connection to a file.
4. The handler keeps a reference to that Thread, associating it with a UID it created and returns the UID to the XML-RPC server.
5. The XML-RPC server returns the UID to the client.

Then, at some point in the future.

1. A request comes into the server from Client A, the request is to get the next 5 records in the file it requested be opened earlier and it passes the UID it received along with the request.
2. The XML-RPC server creates Thread B, (it would have reused Thread A but it's busy with a different, irrelevent request with a different client) and Thread B calls the handler again to execute nextRecords(uid, 5) or something similar.
3. Now the handler is in Thread B executing nextRecords. It needs to execute the actual searching of the file from Thread 1 (the thread the file opening was created from) and then return the values. How do I execute this code in Thread 1? I can get a reference to the Thread, I can get a reference to the ClaimFileIO object that holds the connection, but how do I execute code in ClaimFileIO to get next 5 records FROM the Thread. It's existing, I can't just create a new one and do start(), it has to be same Thread it was opened with.

Here's a solution (I hope, I haven't finished and tested it yet) that I came up with:

public final class ClaimHandler {
    // Holds the ClaimThreads and associates a UID with them
    private Hashtable<String, ClaimThread> threadPool = new Hashtable<String, ClaimThread>();
 
    public ClaimHandler() {
        // nothing to do
    }
 
    public void openConnection() {
        ClaimThread thread = new ClaimThread();
        String uid = UIDFactory.getUIDAsString()
        new Thread("ClaimThread(" + uid + ")", thread).start();
        thread.openConnection();
        threadPool.put(uid, thread);
    }
 
    public void getNext(int records) {
        ClaimThread thread = threadPool.get(uid);
        thread.getNext(records);
    }
 
    public void closeConnection() {
        ClaimThread thread = threadPool.get(uid);
        thread.closeConnection();
        threadPool.remove(uid);
    }
}
 
public interface Command {
    public void execute();
}
 
public ClaimThread implements Runnable {
    // Holds commands that need to be executed
    private Stack<Command> commandQueue = new Stack<Command>();
    private ClaimFileIO claimFile = ClaimFileIO.createInstanceOf();
 
    public ClaimThread() {
        // nothing to do
    }
 
    public void run() {
        while (claimFile != null) {
            if (!commandQueue.empty()) {
                commandQueue.pop().execute();
            }
        }
    }
 
    public void openConnection() {
        Command com = new Command() {
            public void execute() {
                claimFile.openIO();
            }
        };
        commandQueue.push(com);
    }
 
    public void closeConnection() {
        Command com = new Command() {
            public void execute() {
                claimFile.close();
            }
        };
        commandQueue.push(com);
        claimFile = null;
    }
 
    public void getNext(int records) {
        Command com = new Command() {
            public void execute() {
                ... get the next records ...
            }
        };
        commandQueue.push(com);
    }
}

This is missing a couple things. First, it's not returning anything. Second, it's not waiting for the Command to execute before it returns. I can think of a few ways of accomplishing that, which I did not include in this example because I want to know if my methodology of a "Command Queue" is good to begin with.

Also, I'm aware that this is probably not thread-safe. I can see a few places where synchronization is a must, though if you feel ambitious enough to point them all out (in case I'm missing something) that'd be awesome.

So can anyone PLEASE give me some idea of how good or bad my methodology is and suggest a better methodology if mine is bad?

[ October 13, 2005: Message edited by: Ken Blair ]
[ October 13, 2005: Message edited by: Ken Blair ]

[Stan James]

You're well on track with the "queue of commands" concept. There's a nice thread pool in JDK5 that does just this with a BlockingQueue. If you're on an older JVM the Apache Commons thread pool is good. It's a neat reading experience even if you are on JDK5.

Here's a tiny example using JDK5:

    private void acceptClientRequests()
    {
        executor = Executors.newCachedThreadPool( this );
        while (mRunning)
        {
            try
            {
                executor.execute( new MessageHandler( mServerSocket.accept() ) );
            }
            catch (IOException e)
            {
                if (mRunning)
                    throw new WikiException( "HTTPServer.acceptClientRequests Error accepting request: " + e, e );
            }
        }
    }

Here the MessageHandler is the command, implementing run() to do the work for each new incoming request.

Hope that helps!
[ October 13, 2005: Message edited by: Stan James ]

[Ken Blair]

I thought in a "thread pool" you had no control over which Thread you ran a piece of code with. I named my variable "threadPool" but to my understanding it's really not a pool at all, it was just a convenient name. Am I wrong?

[Stan James]

In JDK 5 you could real simply build a producer thread that puts commands into a blocking queue and a consumer thread that gets and executes. The blocking queue would take care of a lot of details you've been getting into. It efficiently blocks your consumer thread when the queue is empty and releases it when work comes available. The main thing a thread pool does differently is manage a number of threads for you. I may have gone more complex than you need when I went into pools. Look at BlockingQueue doc and see what you think.

[Ken Blair]

Thanks I'll look at that. I just need to be careful since the issue with accessing COBOL files seems to add another level of complexity by having to track which specific Thread they're asking for and when that Thread has closed it's connection and could potentially open a new one.

It's funny how things work though. I spent hours looking for a solution on the internet to find nothing useful. I come up with my own solution and then find a myriad of articles that would have led me straight to what I came up on my own. Why couldn't I have found things like Goetz's article a few days ago?

[Mr. C Lamont Gilbert]

why does it matter which thread runs the code? That should be abstract.

Command pattern or strategy is good way to allow a thread to run different code at different times.

[Ken Blair]

Originally posted by Mr. C Lamont Gilbert:
why does it matter which thread runs the code? That should be abstract.

Command pattern or strategy is good way to allow a thread to run different code at different times.

See the OP;

Originally posted by Ken Blair:
I'm working with COBOL files and in order to open a new connection to a file it must be done in a new Thread. The fun part comes when I have to keep that Thread alive and wait for further requests to be made on that file.

That's why. Opening a connection without it being a new thread will fail. Trying to use that connection from a thread other than the one it was created from will fail. That part is out of my hands so there's nothing I can do about it.

[Stan James]

I'm curious about your protocol to COBOL ... is that a mainframe call over ECI or JCA or something?

Sounds like you want to start a consumer thread that holds the connection and can execute a series of calls over the connection. I can imagine something like:

BlockingQueue q = new BlockingQueue();
CobolService s = new CobolService(q);
s.start();
q.put( new ConnectCommand() );
q.put( new GetDataCommand() );
q.put( new PutDataCommand() );
q.put( new DisconnectCommand() ); // tells s to close conn and end thread

Do you need to interact more closely with this thing, maybe know when each command is done?

CobolService s = new CobolService(q, this); // s needs this to notify()
q.start();
q.put( new GetDataCommand() );
wait();
String x = s.getResults();
q.put( new PutDataCommand( newData ));
wait();

If this is making sense, look into Future and Callable for much cleaner ways of getting the results back.
[ October 16, 2005: Message edited by: Stan James ]

[Ken Blair]

We use Micro Focus COBOL, apparently Micro Focus has a Java package that facilitates calling COBOL from Java and that's what's being used. I'm just now starting to learn COBOL and I don't know the details. All I really know is is that "cobcall" (which is part of MF's Java stuff) is being used to execute a COBOL program that was written specifically to be called by Java to access indexed files.

Right now I'm effectively waiting for the Command to finish execution then returning with the results. I've switched to a BlockinQueue instead of a Stack and seen a 10x improvement already. I've been reading up on the new JDK 5.0 goodies for this and I'll certainly look at Future and Callable, thanks for the tip.

What I'm basically looking at doing now is having a common pool of "worker" threads that can execute "commands" passed to them. Each thread will be able to handle an indefinite number of unique connections (i.e. one to an SC01.DAT file, one to an SP01.DAT file, but never two to the same). The fun part is keeping track of which threads have which types of connections open. If a thread has a claim connection open, I can't use it for a new claim connection but I can use it for any other type. I'm looking at either associating a flag with each worker thread that determines whether or not it can be used for specific type of connection, or having seperate 'pools' for each type that use the same threads but each pool only holds the threads that are available for that type of connection. Not really sure what to do here with that.

[Stan James]

I'd be tempted to make a queue per connection so all the commands in that queue are intended for the same collection. You could tune one or two or ten threads pulling from each queue depending on the work volume.

Sounds like you're well on the way, though. Let us know how it all winds up!

[Ken Blair]

That's where I started, a queue per connection and each connection having it's own thread. However, seeing as there's many different types of connection I fear stability issues under heavy load. If each connection has it's own thread, then I could potentially end up with thousands of threads under a heavy load. Since each thread can actually handle multiple connections as long as there's only one connection per file it doesn't make sense to do a one thread to one connection model to me. I'm thinking making it a one thread to many connections will greatly reduce the chances of it dying under heavy load.

The problem is I still need to provide a way of identifying which connection a client is associated with and a way of managing which threads can have open connections to which files so I can find a thread that doesn't have a connection open of the same type I need. So here's what I've done today.

1. Created a ThreadWorker class that extends thread and provides an execute(Runnable) method so that Runnables may be passed to it for execution. It uses a BlockingQueue for this and will destroy itself if there's no activity in x seconds.

2. Created a ConnectionManager interface. The purpose of the ConnectionManager is to, well, manage a connection. Each connection 'type', which for our purposes is basicaly a different file, has an implemenation of ConnectionManager associated with it. When a new connection of that type is needed the implemented ConnectionManager gets a new worker from ThreadManager (a worker is a ThreadWorker) and associates it with a unique ID. All future calls made with the unique ID will be executed in that worker until it is closed or the connection times out, in which case the worker is returned to the ThreadManager.

3. Created a ThreadManager interface. The purpose of the ThreadManager being to provide similar functionality to a thread pool. A ConnectionManager registers itself with the ThreadManager at which point the ThreadManager starts keeping track of which of it's workers in it's pool are 'available' for that connection type. How it does this is implementation dependent. When a ConnectionManager requests a new worker it finds one that is 'available' for that connection type and returns it, creating a new one if necessary and making the returned worker 'unavailable' for that connection type. When the ConnectionManager returns the worker it's made 'available' for that connectiont ype again.

The overall design seems satisfactory to me, it allows for a pool of threads and tracks which threads can be used for which types at any given time returning a new thread that can be used for a specific type on request. I'm not sure how to implement it though. I made two different implementations already and I don't like either.

The first is a "PoolingThreadManager" that creates a new queue for each ConnectionManager that registers. It puts all of it's workers in this new queue. When that ConnectionManager requests a new worker it gets the head, when it returns one it adds it to the tail. This results in what could be described as a pool for each ConnectionManager that has the workers that are 'available' for that kind of connection. I don't like this for a few reasons. First, it means I've got to store a different queue for every damn connection type, eating up memory. Second, it means I have to use a map to map each connection type to the queue backing it, more memory eaten up. Third, if a new worker is necessary it means I ahve to iterate over the map's values (which are queues) adding it to each one.

The second is a "FlaggingThreadManager" that has a map that links each worker to a set of flags (2 bytes). When a ConnectionManager is registered, it assigns a new value to that ConnectionManager. When a new worker is requested, it gets that ConnectionManager's associated value and iterates through the map checking each thread to see if the flag is set for that ConncetionManager. It returns the first one it finds unset, setting the flag (which marks it unavailable). Upon returning the worker it unsets that flag again (which marks it available). I don't like this because for every worker that's requested I have to iterate over the map with the threads & flags. Since I have to unset the flag when the thread returns, it becomes necessary to maintain two maps, one with the flags as keys and one with the workers as keys.

I guess the first one will probably be faster most of the time while the second one will require less memory, especially if there's a lot of ConnectionManager's registered. Neither seem good to me, I don't know what else to do though. Perhaps the design is flawed to begin with, but with my requirements I think it's about as good as can be expected.

I looked at Executor and ExecutorService, but they don't really seem useable or appropriate given the tight control I need over the threads. They're supposed to decouple what needs running from how it's run, which is exactly what I can't have. I must be able to specify which thread code is run in, it's totally out of my hands. I am using Callable and FutureTask as was suggested, those are perfect for working with my ThreadWorker to execute the code and return the result. It's managing the open threads that's being a pain.

I'm debating posting my complete source code. If someone wants to look it over and comment I'll be happy to, it doesn't reveal any trade secrets or anything that will get me into trouble. Seems like there's very little out there on something like this, I guess this kind of tight coupling being necessary is rare. Anyway, thanks for the help.

[Ken Blair]

Just in case you're wondering why I bother with a ThreadWorker when there's Executor:

From the API:
This interface provides a way of decoupling task submission from the mechanics of how each task will be run, including details of thread use, scheduling, etc

Since I need tight control over the mechanics of how eah task will be run, specifically the details of thread use, it seems inappropriate and misleading to use even though I could. Agree? Disagree?