I've thrown everything I can at this, and I can't get it to lock or crash. My hope is that I have applied the principles correctly. I write client apps in JavaScript, and this is only the 3rd .NET class I've ever written (using JScript), so a topic as advanced as this really needs a once-over by some experienced developers.
This is a wrapper for the client class I wrote for a custom client/server exchange. The purpose is to allow asynchronous usage via threading. Originally I wanted to use the thread pool since each task shouldn't last very long, but JScript.NET cannot create delegates. The supposed solutions to this found here and here won't compile. So instead I set out to create my own threads. It's basically just a producer/consumer paradigm.
Rather than create many threads for processing calls, I opted to use a single long-lasting thread and pulse/wait to handle the calls. This is mainly because the server operates in a single thread anyway, so it can't accept multiple connections. This worked well, but on refreshing my .hta file, the class tried to reuse the same thread. Correcting that would require code to run on the unload event. In my experience that event doesn't always fire, but I'm not really sure why.
I thought it best to just use a separate thread for each consumer. I thought that went well, until I fired a test load of 500k calls and realized too late that I hadn't thought to throttle the thread generation. I fixed that by allowing each consumer to determine if another consumer is required after it has finished processing the current queue item.
I'm hoping for some feedback on if I made good choices, as well as if the output is a reasonable way to accomplish my goal.
Notes:
My comments are right-aligned, so you might not see them in the code viewport (I'm not good at reading code when it's interspersed with comments), but I have included a brief explanation of what I was trying to do, and why I did it at the head of each function.
dataLockis used insendMessage()andconsumeMessage()and should provide locking formessageQueueandmessageIDreceivedLockis used inqueryMessage()andconsumeMessage()and should provide locking forreceivedMessageconnectionLockis used inConnected()andconnectionThread()and should provide locking forisConnectedConnected()andqueryMessage()should try not to block for very long
I theorize that 1 or 2 threads might be created, albeit short lived, if the 5 threads are generated and consumer decides to create another thread. But they won't have anything to act upon and just exit harmlessly (if they exist). I don't see any un-closed threads with Process Explorer, so I'm not terribly worried. It just smells funny.
If I get this right I hope to add multithreading to the servers-reading mechanism as well.
import System;
import System.Threading;
import BlazinWeb.Comm
package BlazinWeb.Async.Comm {
class Client {
static var
isConnected : boolean = false ,
messageID : int = 0 ,
messageQueue : Array = [] ,
consumerThreadQueue : Array = [] ,
workingThreadsQueue : Array = [] ,
receivedMessage : Object = {} ,
comm : Object = new BlazinWeb.Comm.Client ,
// our locks
dataLock : Object = {} ,
receivedLock : Object = {} ,
connectionLock : Object = {};
public function init() : void {
// a logfile value of '' ensures that no file will be written to.
console.logFile = 'logFile.txt';
console.clear();
messageQueue = [] , // reset these to default values. refreshing the .hta reuses this instance for whatever dumb reason, so to make sure everything plays nice we put these back
messageID = 0 ,
receivedMessage = {};
var connect : Thread = new Thread(ThreadStart(connectThread));
connect.IsBackground = true;
connect.Start();
}
public function Connected() : boolean {
//
// using connectionLock our application can validate that we have in fact
// established a connection, and it is ok to start transmitting. At least
// this one is pretty straight forward; it's only used here and in
// connectThread, and in both we have just one thing to accomplish.
//
var connected : boolean = false;
Monitor.TryEnter(connectionLock); // use TryEnter to kick back as quick as possible
if (Monitor.IsEntered(connectionLock)) {
try {
connected = isConnected;
}
finally {
Monitor.Exit(connectionLock);
}
}
return connected;
}
private static function connectThread() : void { // console.log('enter connectThread thread')
//
// The comm handshake can block for upwards of 5 seconds or so logging in
// and getting a sessionID. So it gets its own thread; that way we can
// return control back to our application immediately. The user can't really
// do anything of value while this is going on, but at least this way the
// UI doesn't lock up and appear as though the application has frozen.
//
comm.SetChannelName('channel 2');
comm.SetIP('10.10.1.10');
comm.SetPort(8888);
Monitor.Enter(connectionLock); // block until we get access
// console.log('enter connectionLock')
try {
isConnected = comm.Connect();
}
finally {
Monitor.Exit(connectionLock); // console.log('exit connectThread');
}
}
public function sendMessage(message : String) : int { // add message to messageQueue
//
// using dataLock we take control of the messageQueue and messageID. Even though
// we don't want to block the application for too long, we do need to stay here
// until we get access to that lock; so let's make sure that noone else gets too
// greedy with it.
//
// While locked: get a local copy of the messageID so we have something to return
// to our application that is non volatile. Add the incoming message to the
// messageQueue, along with it's ID for later lookup.
//
// if the length of the queue is <5 go ahead and spin up a new thread so that we
// have a few waiting when there are a lot of requests. otherwise- don't make a
// consumer thread. the consumer will decide if a new thread is needed at the end
// of its execution. in this way we have throttled thread creation to 5 concurrent
// consumer threads at any given moment.
//
var createThread : boolean = false;
var currentID : int = 0;
var consume : Thread = new Thread(ThreadStart(consumeMessage)); // we have one consumer thread for each message produced
Monitor.Enter(dataLock);
try {
currentID = messageID;
if (messageQueue.length<5) // allow a few threads to be generated and running concurrently.only one at a time can access the synchronous Comm.Client class, but at least this way we don't need to wait for a new thread to spin up.
createThread = true; // although, this is technically only if sendMessage is called >5 times in rapid succession...
messageQueue.push([messageID , message]);
++messageID;
}
finally {
Monitor.Exit(dataLock);
}
if (createThread) {
consume.IsBackground = true;
consume.Start();
}
return currentID;
}
public function queryMessage(id : String) : String {
//
// using receivedLock we take control of receivedMessage and check whether
// the information we are looking for exists. The application can poll this
// as many times as it wants waiting for a response.
//
// Once we make a local copy of the response to relay to the application we
// want to delete it from our lookup table, otherwise we would build up a
// lot of useless information.
//
var message : String = '';
Monitor.TryEnter(receivedLock); // use TryEnter to kick back as quick as possible
if (Monitor.IsEntered(receivedLock)) {
try {
if (receivedMessage && receivedMessage[id]) {
message = receivedMessage[id];
delete receivedMessage[id]; // we are finished with this information.
}
else {
//console.log('not ready yet');
}
}
finally {
Monitor.Exit(receivedLock);
}
}
else {} // couldn't get a lock this time. better luck next time
return message;
}
private static function consumeMessage() : void { // only one consumer thread can be active at a time.
//
// Ok, here's the tricky one. In order to consume the message we need to
// lockdown the messageQueue and messageID and receivedMessage lookup table.
// As mentioned in sendMessage we don't want to get greedy with the
// messageQueue, because doing so will block our producer and thus our UI.
// So, we want to get a local copy of the next item in the queue, and release
// it before transmitting. When we get our reply- we can safely store it and
// re-lock the queue to remove this item.
//
// The transaction is only considered successful if we dequeue this item AND
// store a reply in the lookup table. For this reason, we use dataLock and
// receivedLock at the same time. We end up with a lot of ugly try/finally's
// embedded within eachother, but it all seems to jive.
//
// If our transmission times out it will throw an exception, otherwise it
// will always return a string (because we sent a string, but there is an
// override where we could send and object- in which case it would always
// return an object if it doesn't throw a timeout exception). If an exception
// is thrown, we don't bother locking our queue or lookup table, and the
// next time around we will try the same item again. Anything that is not
// an exception will contain a lastReceived.asString that our application
// knows how to handle. If we continue to get timed out- our server is no
// longer reachable, in which case our sessionID is no longer valid, and
// we basically need to scrap all this and start over at the application
// level anyway.
//
// Lastly, if there are any more messages in our queue after this item is
// removed go ahead and spin up another consumer thread- start this whole
// game over again. Since this step is bypassed when we get timed out, we
// need to automatically spin up another consumer in that event. If not
// we still have messages in the queue (at least the one that just failed)
// but no threads to consume them.
//
var createThread : boolean = false;
var next : Array = null;
var consume : Thread = new Thread(ThreadStart(consumeMessage));
Monitor.Enter(connectionLock); // block until we get access
try { // wrap our entire method with a try. if anything goes awry we will still exit connectionLock allowing the next thread a chance to use the channel
Monitor.Enter(dataLock); // block until we get access
try { // real quick get the next message in our queue
if (messageQueue[0]) { // console.log('found ' + messageQueue[1])
next = messageQueue[0];
}
}
finally {
Monitor.Exit(dataLock); // let go before we go across the internet so that our producer can keep producing while we are gone
}
if (next) { // dont bother if our queue failed to produce a 'next' array
try {
comm.sendMessage(next[1]); // off we go... weeeeeee! (sendMessage is a blocking call across tcp/ip)
}
catch(e) {
console.log('timeout'); // the only exception thrown by sendMessage is 'Transmission Timed Out'
}
if (comm.lastReceived.asString && comm.lastReceived.asString !== '{}') { // console.log('got something back')
Monitor.Enter(dataLock); // block until we get access
try {
Monitor.Enter(receivedLock); // block until we get access
try {
messageQueue.shift(); // console.log(messageQueue.length)
if (messageQueue.length>0)
createThread = true;
receivedMessage[next[0]] = comm.lastReceived.asString; // console.log(comm.lastReceived.asString)
}
finally {
Monitor.Exit(receivedLock);
}
}
finally {
Monitor.Exit(dataLock);
}
}
else { // the comm threw a timeout error
createThread = true; // replace the thread we lost, or else if the server hangs enough to reduce thread count to 0, the component will lose the ability to create new ones
console.log('ruh roh'); // right now we do nothing. the message that timed out will try again at our next iteration.
// maybe later we can set the received message to 'request failed' and allow the calling application decide if it wants to try again.
}
}
}
finally {
Monitor.Exit(connectionLock);
}
if (createThread) {
consume.IsBackground = true;
consume.Start();
}
}
}
}
