[java] Rolling (Shared) Queue

[Chavo]

So apparently I'm one of those strange people that still has a legitimate reason to do battle.net related programming.

I wrote my own bot with some very unique core features.  One of the biggest features is the ability to handle multiple connections to battle.net inside one process.  This lets me send outgoing messages on a 'rolling' or shared system.

For example, in any bot that I know of, if you wanted to send 10 long messages, it would probably take you a while to avoid flooding off.  However, with a rolling Queue setup, you can divide those messages over a number of connections so that they are displayed much faster while still avoiding flooding off.

I've written, tested, debugged, and racked my brian for improvements on the code I'm using and it works pretty well.  However, sometimes it will display messages in the wrong order (for example, if I enqueue'd messages 1,2,3,4,5,6,7,8,9,10 it might display them as 1,2,3,4,5,7,8.6,10,9).  The more messages in the queue(s), the more likely they will be displayed out of order. 

iago, this is the code that I was talking to you about a while back that we both forgot I still wanted help with

I'm looking for any ideas or suggestions to help elliminate the problem or any insight to why it is occuring because I've tried to anticipate and prevent it in my code. Without further ado:

Queue.java

Code: [Select]

package connect;

import java.util.Timer;
import java.util.TimerTask;
import java.util.PriorityQueue;

import util.BNetPacket;
import util.QueuePacket;

/***
* Queue manages timing of outgoing data for a given PacketThread connection.
* Numbers and algorithm for computing proper delays inspired by iago's JavaOp2 code (www.javaop.com)
*
*@author unTactical
*/
public class Queue {

private final PriorityQueue<QueuePacket> queue = new PriorityQueue<QueuePacket>();
private final Timer timer = new Timer();

private TimerTask current;
private PacketThread conn;
    private boolean queueReady = true;
    private boolean enabled = false;
    private long lastSent = System.currentTimeMillis();

//TODO: make these customizable
    int packetCost = 250;
    int byteCost = 17;
    int byteOverThresholdCost = 17;
    int thresholdBytes = 65;
    int maxCredits = 800;
    int creditRate = 5;
    int credits = 800;   
   
    public Queue(PacketThread connection)
    {
    conn = connection;
    }
   
    public Queue(String DONOTUSESTHIS_FORTESTINGONLY) // for testing with QueueManager main function
    {
    }
   
    /** Supports disabling of the queue temporarily (useful for faster logon)
     *
     * @param status - enable/disable
     */
    public void enable(boolean status)
    {
    enabled = status;
    }
   
    /** Clear the queue if it gets too full  */
    public synchronized void clear()
    {
    queue.clear();
    }
   
    /** Primary function of Queue, schedules a packet to be sent, ordered by Priority and time scheduled
     *
     * @param data - packet to send to battle.net
     * @param priority - higher means it will be sent sooner
     */
    public synchronized void send(BNetPacket data, int priority)
    {
    QueuePacket q = new QueuePacket(data,priority);
    queue.add(q);
    if(queueReady)
    {
    long delay = getDelay(data.getBytes());
    timer.schedule(new QueueTask(),delay);
    queueReady = false;
    }
    }
   
    /** Tells us if this Queue is ready to send a message immediately or if it will have to be scheduled
     *
     * @return true if message can be sent immediately, false if we must schedule it
     */
    public synchronized boolean isReady()
    {
    return queueReady;
    }
   
    /** Lets us predict how long it will be before a new message can be sent based on this queue's status.
     * This is the core function that allows us to use a rolling queue system.
     *
     * @param priority - only consider packets in the queue matching or exceeding a given priority
     * @return amount to wait
     */
    public synchronized long getWait(int priority)
    {
    QueuePacket qpacket;
    updateCredits();
    int wait = credits;
   
    if(queue.size() > 0)
    {
    Object[] elements = queue.toArray();
    int tempCredits = credits;
    for(int i=0;i<elements.length;i++)
    {
    qpacket = (QueuePacket) elements[i];
    if(qpacket.getPriority() >= priority)
    getDelay(qpacket.getData().getBytes());
    }
    wait = credits;
    credits = tempCredits;
    }
   
    return wait;
    }
   
    /** Does the math for figuring out how long to wait after a message is sent
     * If you call this method, it adjusts the queue's status so if you don't really want to send the message,
     * make sure you return credits to its previous value.
     * I borrowed and tweaked this from iago's code.
     *
     * @param bytes - the longer the message, the more we need to wait.  This could be replaced with a length variable
     * @return time to wait after sending a message of length specified
     */
    private long getDelay(byte[] bytes)
    {   
    if(!enabled)
    return 0;
   
    // Add the credits for the elapsed time
    updateCredits();
        lastSent = System.currentTimeMillis();

        // Get the packet's "cost"
        int thisByteDelay = byteCost;       
        if(bytes.length > thresholdBytes)
            byteCost = byteOverThresholdCost;
        int thisPacketCost = packetCost + (thisByteDelay * bytes.length);

        // Check how long this packet will have to wait
        int requiredDelay = 0;
        // If we can't "afford" the packet, figure out how much time we'll have to wait
        if(credits < 0)
            requiredDelay = -credits * creditRate;

        // Deduct this packet from the credits
        credits -= thisPacketCost;
        return requiredDelay;
    }
   
    /** Since credits doesn't change on its own, we need a way to ensure its status is correct
     * when determining wait time.
     */
    private void updateCredits()
    {
        if(credits < maxCredits)
        {
            credits += (System.currentTimeMillis() - lastSent) / creditRate;
           
            if(credits > maxCredits)
                credits = maxCredits;
        }       
    }
   
    /** Waits after a message is sent and sends the next message in line, then schedules the next wait
     * If there is no next message to send, it sets the queue to a 'ready' status.
     *
     * @author unTactical
     */
    private class QueueTask extends TimerTask
    {
    QueuePacket data;   
   
    public void run()
        {
        if(!queue.isEmpty())
        {
        // send the message and schedule the next one
        data = queue.remove();
        byte[] tosend = data.getData().getBytes();
        conn.send(tosend);
        long delay = getDelay(tosend);
        timer.schedule(current = new QueueTask(),delay);

        queueReady = false;
        }
        else
        queueReady = true;
        }
    }
}
QueueManager.java

Code: [Select]

package _main;

import java.util.HashMap;

import events.OutChatCommand;
import connect.Queue;
import plugins.Logger;
import util.BNetPacket;

/** Central point for all outgoing data to battle.net
 *  - IndividalQueueArray: add to a specific bot
 *  - AllQueueArray: add to all bots
 *  - OpOnlyArray: add to the most available bot with ops
 *  - NonOpArray: add to the most available both without ops
 *
 * @author unTactical
 */
public class QueueManager {

private static HashMap <String, Queue> all = new HashMap<String, Queue>();
private static HashMap <String, Queue> ops = new HashMap<String, Queue>();
private static HashMap <String, Queue> nonops = new HashMap<String, Queue>();

private QueueManager() {}

/** Lets us disable flood protection for a given Queue so that messages are sent immediately
*
* @param id - the name of the queue to enable/disable
* @param status - true to enable flood protection, false to disable
*/
public static void enable(String id, boolean status)
{
all.get(id).enable(status);
}

/** Add a Queue to the Manager
*
* @param id - name of the Queue
* @param q - Queue to manage
* @param hasOps - whether this bot handles 'operator' or non 'operator' messages by default
*/
public synchronized static void addQueue(String id, Queue q, boolean hasOps)
{
all.put(id,q);
if(hasOps)
ops.put(id,q);
else
nonops.put(id, q);
}

/** Remove a queue no longer in use
* This usually happens when we reset a connection.
*
* @param id - name of the Queue to remove
*/
public synchronized static void removeQueue(String id)
{
all.remove(id);
ops.remove(id);
nonops.remove(id);
}

/** Allows us to change whether a given Queue handles ops or non ops messages
*
* @param id - name of the Queue
* @param hasOps - true to handle 'operator' messages, false to handle non 'operator' messages by default
*/
public synchronized static void updateOps(String id, boolean hasOps)
{
if(hasOps)
{
nonops.remove(id);
if(!ops.containsKey(id))
ops.put(id, all.get(id));
}
else
{
ops.remove(id);
if(!nonops.containsKey(id))
nonops.put(id, all.get(id));
}
}

/** Allows us to change whether a given Queue handles ops or non ops messages
*
* @param id - int ID of the Queue to be looked up
* @param hasOps - true to handle 'operator' messages, false to handle non 'operator' messages by default
*/
public synchronized static void updateOps(int id, boolean hasOps)
{
updateOps(ConnectionManager.getInstance().getConnection(id).getName(), hasOps);
}

/** Remove all Queues
* Usually done when there is a connection error and all Queues must be reset
*/
public static void clearAll()
{
Object[] queues = all.values().toArray();
for(int i=0;i<queues.length;i++)
{
((Queue) queues[i]).clear();
}
}

/** Sends a message on all Queues in this Manager
*
* @param data - packet to send to battle.net
* @param priority - order in line, greatest first
*/
public static void sendAll(BNetPacket data, int priority)
{
Object[] queues = all.values().toArray();
for(int i=0;i<queues.length;i++)
{
((Queue) queues[i]).send(data,priority);
}
}

/** Send a message on a specific Queue
* Primarily done when a packet must be returned on the same Queue it was received (ie logon)
*
* @param data - packet to send to battle.net
* @param priority - order in line, greatest first
* @param id - name of the Queue we want to send the packet on
*/
public static void sendByName(BNetPacket data, int priority, String id)
{
Queue sender = all.get(id);
if(sender != null)
sender.send(data, priority);
else
Logger.logConsoleText("Sender id " + id + " is null!");
}

/** Send a message on a specific Queue
* Primarily done when a packet must be returned on the same Queue it was received (ie logon)
*
* @param data - packet to send to battle.net
* @param priority - order in line, greatest first
* @param id - int ID of the Queue to lookup
*/
public static void sendByID(BNetPacket data, int priority, int ID)
{
sendByName(data, priority, ConnectionManager.getInstance().getConnection(ID).getName());
}

/** Send a non 'operator' message. 
* The message will be sent on the first queue that is ready. If no queue is ready, it is added to the Queue
* that will be ready to send it soonest
*
* @param data - packet to send to battle.net
* @param priority - order in line, greatest first
*/
public static void sendOps(BNetPacket data, int priority)
{
Object[] queues = ops.values().toArray();
sendSpecified(queues,data,priority);
}

/** Send an 'operator' message.
* The message will be sent on the first queue that is ready. If no queue is ready, it is added to the Queue
* that will be ready to send it soonest
*
* @param data - packet to send to battle.net
* @param priority - order in line, greatest first
*/
public static void sendNonOps(BNetPacket data, int priority)
{
Object[] queues;
if(!nonops.isEmpty())
queues = nonops.values().toArray();
else
queues = ops.values().toArray(); // If there is no non-ops Queue available, it will be sent on an op Queue instead.

sendSpecified(queues,data,priority);
}

/** Checks all of the Queues availability and chooses (ideally) the first Queue that is ready to send the message.
* If no Queue is ready, it determines which Queue will be ready to send the message soonest and adds it to that one.
* (Ideally)
*
* @param queues - spefies whether we are checking the Ops Queue or the NonOps Queue
* @param data - packet to send to battle.net
* @param priority - a message of a higher priority is sent before messages of lower priority, so only check those relevant
*/
private static synchronized void sendSpecified(Object[] queues, BNetPacket data, int priority)
{
long highCredits = -999999;
int lowID = -1;
Queue current;

for(int i=0;i<queues.length;i++)
{
current = ((Queue) queues[i]);
long thisWait = current.getWait(priority);

if(thisWait == 800)
{
current.send(data, priority);
try {Thread.sleep(500); } catch(Exception e) {};
return;
}
else if(thisWait > highCredits)
{
highCredits = thisWait;
lowID = i;
}
}

if(lowID == -1) // this will happen if the queue is empty (ie a op message is requested but no op queue exists)
Logger.logConsoleText("Message not sent! lowID invalid!!");
else
((Queue) queues[lowID]).send(data, priority);

try {Thread.sleep(500); } catch(Exception e) {}; // wait a short time to account for non-0 travel time across internet
}

public static void main(String[] args)
{
addQueue("zero",new Queue(""),false);
addQueue("one",new Queue(""),false);
System.out.println("test1");
sendNonOps(new OutChatCommand(-1,"test1").getBNetPacket(),0);
System.out.println("test2");
sendNonOps(new OutChatCommand(-1,"test2").getBNetPacket(),0);
System.out.println("test3");
sendNonOps(new OutChatCommand(-1,"test3").getBNetPacket(),0);
System.out.println("test4");
sendNonOps(new OutChatCommand(-1,"test4").getBNetPacket(),0);
System.out.println("test5");
sendNonOps(new OutChatCommand(-1,"test5").getBNetPacket(),0);
System.out.println("test6");
sendNonOps(new OutChatCommand(-1,"test6").getBNetPacket(),0);
System.out.println("test7");
sendNonOps(new OutChatCommand(-1,"test7").getBNetPacket(),0);
System.out.println("test8");
sendNonOps(new OutChatCommand(-1,"test8").getBNetPacket(),0);
System.out.println("test9");
sendNonOps(new OutChatCommand(-1,"test9").getBNetPacket(),0);
System.out.println("test10");
sendNonOps(new OutChatCommand(-1,"test10").getBNetPacket(),0);
}
}

[Chavo]

Do I need to do a better job of explaining or is it just too complex for anyone to have the time to look at it at the moment?

Don't tell me I cleaned up all my comments for nothing!

[AntiVirus]

I'd help you, but that's above me.

[Chavo]

I don't know about that.  You might not know java well enough to say rewrite it on your own, but I'm not having trouble with java, its the actual ability to get the output ordered correctly that is a dilemna.  I only put the [java] tag on the post because its harder to read if you don't know any java!

That's more math/algorithm intensive than anything.

[nslay]

Hi unTactical,
I wrote a bot in C++ (Chatterbot) last Christmas like yours.  Instead of sharing the send queue between all the bots, I just encapsulated a send queue in each bot object.  On the other hand, a timer queue was shared between all of them.
Timer queue did a few things
Periodic away, periodic sanity check, send queue, and reconnect events

Conceptually, this is how it worked.
I would
1) Read data from config file
2) Initialize each bot
3) run Bot.Startup() which is the heart

It did not use threads

Bot.Startup() would do a few things
It would run FDEnt's Select() function, this would call each bot back that had data waiting on a socket
It would call timer's Exec() function, which would execute all expired pending timers
Then it would grab the remaining time before next timer and then repeat.
Keep in mind, the Select() sleeps for at most the time until the next timer.

The timer class does a callback on a functor, therefore, each bot had to overload the operator()( int )...the argument is the callback event.

If it was a send queue callback, it would call a send queue method to send awaiting data ... if any data was left to be sent, it would queue a timer event again.

Some code snippits:

Code: [Select]

void Bot::operator()( int event ) {
int sz;
char buff[ RECVBUFFSIZE + 1 ];
const char* p;
switch( event ) {
case TCPENT_CONNECT:
#ifdef BOT_DEBUG
Print( "Attempting to connect to %s ...\n", GetConAddr() );
#endif
if ( timeout != zerotime )
timer.Set( this, BOT_TIMEOUT, timeout );
//Print out some info here
break;
case TCPENT_CONNECTED:
#ifdef BOT_DEBUG
Print( "Connected!\nSending login information...\n" );
#endif
if ( !username || !password )
break;
timer.Remove( this, BOT_TIMEOUT );
//Login
TCPENT::Send( "%c%c%s\r\n%s\r\n", 0x03, 0x04, username, password );
if ( home )
TCPENT::Send( "/join %s\r\n", home );
break;
case TCPENT_READ:
#ifdef BOT_DEBUG
Print( "TCPENT_READ\n" );
#endif
sz = Read( buff, RECVBUFFSIZE );
if ( sz <= 0 )
break;
buff[ sz ] = 0;

//Construct inbuff
p = buff;

do {
const char* pbuff = p;
p = parseto( p, "\r\n" );
if ( p != pbuff ) {
if ( !inbuff ) {
buffsz = p - pbuff;
inbuff = new char[ buffsz + 1 ];
strncpy( inbuff, pbuff, buffsz );
}
else {
char* tmp = new char[ buffsz + p - pbuff + 1 ];
strncpy( tmp, inbuff, buffsz );
strncpy( tmp+buffsz, pbuff, p - pbuff );
buffsz += p - pbuff;
delete [] inbuff;
inbuff = tmp;
}

inbuff[ buffsz ] = 0;
}

if ( *p == '\r' || *p == '\n' ) {
Handle();
if ( inbuff )
delete [] inbuff;
inbuff = 0;
buffsz = 0;
}
} while ( *p++ );

break;
case TCPENT_DISCONNECT:
#ifdef BOT_DEBUG
Print( "Disconnected.\n" );
#endif
ClearTmp();
timer.Remove( this ); //Kill all the logtime timers
timer.Set( this, BOT_RECON, recon );
break;
case TCPENT_FAIL:
#ifdef BOT_DEBUG
Print( "Failed to connect. Aborting.\n" );
#endif
break;
case BOT_STATUS:
#ifdef BOT_DEBUG
Print( "BOT_STATUS\n" );
#endif

if ( home && currentchan ) {
if ( strccmp( home, currentchan ) )
Send( SENDQ_MINPRI, "/join %s\n", home );
}

//sendq.Send( DEFSENDPOLICY );

timer.Set( this, BOT_STATUS, stat );
break;
case BOT_SEND:
#ifdef BOT_DEBUG
Print( "BOT_SEND\n" );
#endif
sendq.Send( DEFSENDPOLICY );
if ( !sendq.Empty() )
timer.Set( this, BOT_SEND, sendtime );
break;
case BOT_IDLE:
#ifdef BOT_DEBUG
Print( "BOT_IDLE\n" );
#endif
char runtimebuff[TIMEBUFF];
GetRuntime( runtimebuff, TIMEBUFF );
Send( SENDQ_MINPRI, "/me is a %s %s - Runtime: %s.\n", BOTNAME, VERSION, runtimebuff );
timer.Set( this, BOT_IDLE, idle );
break;
case BOT_RECON:
#ifdef BOT_DEBUG
Print( "BOT_RECON\n" );
#endif
Connect();
break;
case BOT_TIMEOUT:
#ifdef BOT_DEBUG
Print( "BOT_TIMEOUT\n" );
Print( "Connect timeout.\n" );
#endif
Disconnect();
break;
case BOT_LOGGED:
#ifdef BOT_DEBUG
Print( "BOT_LOGGED\n" );
#endif
//Set all the logtime timers
timer.Set( this, BOT_IDLE, idle );
timer.Set( this, BOT_STATUS, stat );
timer.Set( this, BOT_AWAY, zerotime );
break;
case BOT_AWAY:
#ifdef BOT_DEBUG
Print( "BOT_AWAY\n" );
#endif
Send( SENDQ_MINPRI, "/away I'm a %s(%s)!\n", BOTNAME, sysinfo.sysname );
timer.Set( this, BOT_AWAY, away );
break;
}
}
This, above, is the callback infrastructure.

Code: [Select]

//Start the bot
timeval tout = timer.NextSch();
int res;
while ( run && ( res = Select( &tout ) ) >= 0 ) {
timer.Exec();
tout = timer.NextSch();
if ( res > 0 )
printf( "Select() didn't evaluate all descriptors!\n" );
}
if ( res < 0 )
printf( "Select() error!\n" );
timer.Clear();
Here's the heartbeat of the bot.
To be able to bootstrap the bots, on constructor they would queue a connect timer to execute immediately.  That way, as soon as Startup() is called, the timer.Exec() would kick them all into action ... like a series of dominos.

I know in Java there isn't really select() infrastructure, but maybe this will help you make things more compartmentalized rather than sharing a single send queue between all of them.

[Chavo]

Its an interesting setup and I do appreciate you sharing/trying to help but I don't think its as similar as you thought

know in Java there isn't really select() infrastructure, but maybe this will help you make things more compartmentalized rather than sharing a single send queue between all of them.

I'm not using a single Queue, each connection has its own Connection class, which contains a PacketThread for receiving data and a Queue for sending data.  A QueueManager contains references to all of these Queues and does the work of deciding which Queue to put a message in by comparing Queue states. When I say states, I don't mean asleep/awake/working, I mean the values of its members.  The Queue's are not on their own threads, they send the data that needs to be sent to data to the parent Connection which it turn sends it back out on the PacketThread (which encapsulates the Socket).

QueueManager's sole job is to take data and to decide which connection to battle.net should be responsible for outputing that data.  There are a few constraints though.  If the data is specific to a connection, it must be sent on that connection (obviously).  Otherwise, the data should be sent on the connection that will be able to get the message out soonest (following the antiflooding rules that each Queue has internally) but ensuring proper ordering.  The proper ordering is the only thing I'm having trouble with.

As far as I can tell my method seems to be flawless for getting correct ordering, but apparently I'm missing something and that is what I need help with

If you (or anyone else) wants more info on how exactly my bot works I'm more than happy to share.  I plan on releasing full source when I clean up all my comments and a couple standing bugs.

[Joe]

I don't know what you're doing in there (I'm not in the read-and-concentrate mood), but here's what I'd do.

UI adds message to queue.
Queue runs a standard message cooldown check, however, divides the result by n (being the number of active bots) and then rotates through the n bots.

EDIT -
Er..

I'm not using a single Queue, each connection has its own Connection class, which contains a PacketThread for receiving data and a Queue for sending data.

I'd declare these classes, then.
public static class Queue;
public class QueueGlue;
Where each class has it's QueueGlue that does the nessessary mediating between Queue and the SocketThreads, and you have one big static Queue.

[Chavo]

I don't know what you're doing in there (I'm not in the read-and-concentrate mood), but here's what I'd do.

UI adds message to queue.
Queue runs a standard message cooldown check, however, divides the result by n (being the number of active bots) and then rotates through the n bots.

You're assuming (incorrectly) that all messages are of equal length and equal contribution to the anti-flood delaying.

EDIT -
Er..

I'm not using a single Queue, each connection has its own Connection class, which contains a PacketThread for receiving data and a Queue for sending data.

I'd declare these classes, then.
public static class Queue;
public class QueueGlue;
Where each class has it's QueueGlue that does the nessessary mediating between Queue and the SocketThreads, and you have one big static Queue.

If you actually read my code (or even just the comments) you'll very quickly see thats already handled and as I've already said is not the  problem.

[Joe]

You're assuming (incorrectly) that all messages are of equal length and equal contribution to the anti-flood delaying.

Reguardless, the delay time is figured on that precise message. Although it would be possible for you to say something like
1
2
3
4
5
6
7
888888888888888888888888888888888888888888888888888888888888888888888888888
over and over again, and eventually the 8th bot would flood off, it'd be unlikely. You could write an algorithm to balance out which bot gets the longer messages for a nifty challenge, too.

EDIT -
Also, why don't I have a delete button? I don't want to delete this, but it'd be nice to have the option.

[AntiVirus]

What I am about to say could be terribly stupid, but I will say it anyway at the risk of looking like a moron.

Anyway, is there anyway you can assign the message you are taking in to a certain index in an array and then have that array sorted so that the first message is the first one in line.  Then from there just spit out the array?

Is that possible?

[Newby]

You're assuming (incorrectly) that all messages are of equal length and equal contribution to the anti-flood delaying.

...

If you actually read my code (or even just the comments) you'll very quickly see thats already handled and as I've already said is not the  problem.

It's Joe. Don't expect much. He even said he didn't read your post. I was going to comment negatively, but decided against.

[Chavo]

@Joe: I don't even know what you are trying to suggest, but I'm pretty sure its wrong

@Brandon: you have to take anti-flooding delays into account and attempt to predict them when enqueueing messages

@newby: I know, its annoying.

[AntiVirus]

Could you have time intervals when spitting out the array? 

[Chavo]

Could you have time intervals when spitting out the array? 

Yes, but if its too fast, you flood off with long messages or if its too slow you have too much flood protection and make it unnecessary slow.

[AntiVirus]

Can't you time it just right?  I mean, if it's coding in there, then you should only have to worry about the time's once, it's not like the user can change it, unless of course, that's what you want.