aim_txqueue.c

   1 /*
   2  *  aim_txqueue.c
   3  *
   4  * Herein lies all the mangement routines for the transmit (Tx) queue.
   5  *
   6  */
   7
   8 #include <faim/aim.h>
   9
  10 /*
  11  * Allocate a new tx frame.
  12  *
  13  * This is more for looks than anything else.
  14  *
  15  * Right now, that is.  If/when we implement a pool of transmit
  16  * frames, this will become the request-an-unused-frame part.
  17  */
  18 struct command_tx_struct *aim_tx_new(int chan, struct aim_conn_t *conn, int datalen)
  19 {
  20   struct command_tx_struct *new;
  21
  22   if (!conn) {
  23     printf("aim_tx_new: ERROR: no connection specified\n");
  24     return NULL;
  25   }
  26
  27   new = (struct command_tx_struct *)malloc(sizeof(struct command_tx_struct));
  28   if (!new)
  29     return NULL;
  30   memset(new, 0, sizeof(struct command_tx_struct));
  31
  32   new->conn = conn;
  33   new->type = chan;
  34
  35   if(datalen) {
  36     new->data = (u_char *)malloc(datalen);
  37     new->commandlen = datalen;
  38   }
  39
  40   return new;
  41 }
  42
  43 /*
  44  * aim_tx_enqeue__queuebased()
  45  *
  46  * The overall purpose here is to enqueue the passed in command struct
  47  * into the outgoing (tx) queue.  Basically...
  48  *   1) Make a scope-irrelevent copy of the struct
  49  *   2) Lock the struct
  50  *   3) Mark as not-sent-yet
  51  *   4) Enqueue the struct into the list
  52  *   5) Unlock the struct once it's linked in
  53  *   6) Return
  54  *
  55  * Note that this is only used when doing queue-based transmitting;
  56  * that is, when sess->tx_enqueue is set to &aim_tx_enqueue__queuebased.
  57  *
  58  */
  59 int aim_tx_enqueue__queuebased(struct aim_session_t *sess,
  60                                struct command_tx_struct *newpacket)
  61 {
  62   struct command_tx_struct *cur;
  63
  64   if (newpacket->conn == NULL) {
  65       faimdprintf(1, "aim_tx_enqueue: WARNING: enqueueing packet with no connecetion\n");
  66       newpacket->conn = aim_getconn_type(sess, AIM_CONN_TYPE_BOS);
  67   }
  68
  69   /* assign seqnum */
  70   newpacket->seqnum = aim_get_next_txseqnum(newpacket->conn);
  71   /* set some more fields */
  72   newpacket->lock = 1; /* lock */
  73   newpacket->sent = 0; /* not sent yet */
  74   newpacket->next = NULL; /* always last */
  75
  76   /* see overhead note in aim_rxqueue counterpart */
  77   if (sess->queue_outgoing == NULL) {
  78     sess->queue_outgoing = newpacket;
  79   } else {
  80     for (cur = sess->queue_outgoing;
  81          cur->next;
  82          cur = cur->next)
  83       ;
  84     cur->next = newpacket;
  85   }
  86
  87   newpacket->lock = 0; /* unlock so it can be sent */
  88
  89 #if debug == 2
  90   faimdprintf(2, "calling aim_tx_printqueue()\n");
  91   aim_tx_printqueue(sess);
  92   faimdprintf(2, "back from aim_tx_printqueue()\n");
  93 #endif
  94
  95   return 0;
  96 }
  97
  98 /*
  99  * aim_tx_enqueue__immediate()
 100  *
 101  * Parallel to aim_tx_enqueue__queuebased, however, this bypasses
 102  * the whole queue mess when you want immediate writes to happen.
 103  *
 104  * Basically the same as its __queuebased couterpart, however
 105  * instead of doing a list append, it just calls aim_tx_sendframe()
 106  * right here.
 107  *
 108  */
 109 int aim_tx_enqueue__immediate(struct aim_session_t *sess, struct command_tx_struct *newpacket)
 110 {
 111   if (newpacket->conn == NULL) {
 112     faimdprintf(1, "aim_tx_enqueue: ERROR: packet has no connection\n");
 113     if (newpacket->data)
 114       free(newpacket->data);
 115     free(newpacket);
 116     return -1;
 117   }
 118
 119   newpacket->seqnum = aim_get_next_txseqnum(newpacket->conn);
 120
 121   newpacket->lock = 1; /* lock */
 122   newpacket->sent = 0; /* not sent yet */
 123
 124   aim_tx_sendframe(newpacket);
 125
 126   if (newpacket->data)
 127     free(newpacket->data);
 128   free(newpacket);
 129
 130   return 0;
 131 }
 132
 133 /*
 134  *  aim_get_next_txseqnum()
 135  *
 136  *   This increments the tx command count, and returns the seqnum
 137  *   that should be stamped on the next FLAP packet sent.  This is
 138  *   normally called during the final step of packet preparation
 139  *   before enqueuement (in aim_tx_enqueue()).
 140  *
 141  */
 142 u_int aim_get_next_txseqnum(struct aim_conn_t *conn)
 143 {
 144   u_int ret;
 145
 146   faim_mutex_lock(&conn->seqnum_lock);
 147   ret = ++conn->seqnum;
 148   faim_mutex_unlock(&conn->seqnum_lock);
 149   return ret;
 150 }
 151
 152 /*
 153  *  aim_tx_printqueue()
 154  *
 155  *  This is basically for debuging purposes only.  It dumps all the
 156  *  records in the tx queue and their current status.  Very helpful
 157  *  if the queue isn't working quite right.
 158  *
 159  */
 160 #if debug == 2
 161 int aim_tx_printqueue(struct aim_session_t *sess)
 162 {
 163   struct command_tx_struct *cur;
 164
 165   faimdprintf(2, "\ncurrent aim_queue_outgoing...\n");
 166   faimdprintf(2, "\ttype seqnum  len  lock sent\n");
 167
 168   if (sess->queue_outgoing == NULL)
 169     faimdprintf(2, "aim_tx_flushqueue(): queue empty");
 170   else {
 171       for (cur = sess->queue_outgoing; cur; cur = cur->next) {
 172           faimdprintf(2, "\t  %2x   %4x %4x   %1d    %1d\n",
 173                       cur->type, cur->seqnum,
 174                       cur->commandlen, cur->lock,
 175                       cur->sent);
 176       }
 177   }
 178
 179   faimdprintf(2, "\n(done printing queue)\n");
 180
 181   return 0;
 182 }
 183 #endif
 184
 185 /*
 186  *  aim_tx_flushqueue()
 187  *
 188  *  This the function is responsable for putting the queued commands
 189  *  onto the wire.  This function is critical to the operation of
 190  *  the queue and therefore is the most prone to brokenness.  It
 191  *  seems to be working quite well at this point.
 192  *
 193  *  Procedure:
 194  *    1) Traverse the list, only operate on commands that are unlocked
 195  *       and haven't been sent yet.
 196  *    2) Lock the struct
 197  *    3) Allocate a temporary buffer to store the finished, fully
 198  *       processed packet in.
 199  *    4) Build the packet from the command_tx_struct data.
 200  *    5) Write the packet to the socket.
 201  *    6) If success, mark the packet sent, if fail report failure, do NOT
 202  *       mark the packet sent (so it will not get purged and therefore
 203  *       be attempted again on next call).
 204  *    7) Unlock the struct.
 205  *    8) Free the temp buffer
 206  *    9) Step to next struct in list and go back to 1.
 207  *
 208  */
 209 int aim_tx_sendframe(struct command_tx_struct *cur)
 210 {
 211   u_char *curPacket;
 212
 213   if (!cur)
 214     return -1; /* fatal */
 215
 216   cur->lock = 1; /* lock the struct */
 217
 218   /* allocate full-packet buffer */
 219   curPacket = (char *) malloc(cur->commandlen + 6);
 220
 221   /* command byte */
 222   curPacket[0] = 0x2a;
 223
 224   /* type/family byte */
 225   curPacket[1] = cur->type;
 226
 227   /* bytes 3+4: word: FLAP sequence number */
 228   aimutil_put16(curPacket+2, cur->seqnum);
 229
 230   /* bytes 5+6: word: SNAC len */
 231   aimutil_put16(curPacket+4, cur->commandlen);
 232
 233   /* bytes 7 and on: raw: SNAC data */  /* XXX: ye gods! get rid of this! */
 234   memcpy(&(curPacket[6]), cur->data, cur->commandlen);
 235
 236   /* full image of raw packet data now in curPacket */
 237   faim_mutex_lock(&cur->conn->active);
 238   if ( (u_int)write(cur->conn->fd, curPacket, (cur->commandlen + 6)) != (cur->commandlen + 6)) {
 239     faim_mutex_unlock(&cur->conn->active);
 240     printf("\nWARNING: Error in sending packet 0x%4x -- will try again next time\n\n", cur->seqnum);
 241     cur->sent = 0; /* mark it unsent */
 242     return 0; /* bail out -- continuable error */
 243   } else {
 244     faimdprintf(2, "\nSENT 0x%4x\n\n", cur->seqnum);
 245
 246     cur->sent = 1; /* mark the struct as sent */
 247     cur->conn->lastactivity = time(NULL);
 248   }
 249   faim_mutex_unlock(&cur->conn->active);
 250
 251 #if debug > 2
 252   faimdprintf(2, "\nPacket:");
 253   for (i = 0; i < (cur->commandlen + 6); i++) {
 254     if ((i % 8) == 0) {
 255       faimdprintf(2, "\n\t");
 256     }
 257     if (curPacket[i] >= ' ' && curPacket[i]<127) {
 258       faimdprintf(2, "%c=%02x ", curPacket[i], curPacket[i]);
 259     } else {
 260       faimdprintf(2, "0x%2x ", curPacket[i]);
 261     }
 262   }
 263   faimdprintf(2, "\n");
 264 #endif
 265   cur->lock = 0; /* unlock the struct */
 266   free(curPacket); /* free up full-packet buffer */
 267
 268   return 1; /* success */
 269 }
 270
 271 int aim_tx_flushqueue(struct aim_session_t *sess)
 272 {
 273   struct command_tx_struct *cur;
 274
 275 #if debug > 1
 276   int i = 0;
 277 #endif
 278
 279   if (sess->queue_outgoing == NULL)
 280     return 0;
 281
 282   faimdprintf(2, "beginning txflush...\n");
 283   for (cur = sess->queue_outgoing; cur; cur = cur->next) {
 284     /* only process if its unlocked and unsent */
 285     if (!cur->lock && !cur->sent) {
 286
 287       /*
 288        * And now for the meager attempt to force transmit
 289        * latency and avoid missed messages.
 290        */
 291       if ((cur->conn->lastactivity + cur->conn->forcedlatency) >= time(NULL)) {
 292         /* FIXME FIXME -- should be a break! we dont want to block the upper layers */
 293         sleep((cur->conn->lastactivity + cur->conn->forcedlatency) - time(NULL));
 294       }
 295
 296       if (aim_tx_sendframe(cur) == -1)
 297         break;
 298     }
 299   }
 300
 301   /* purge sent commands from queue */
 302   aim_tx_purgequeue(sess);
 303
 304   return 0;
 305 }
 306
 307 /*
 308  *  aim_tx_purgequeue()
 309  *
 310  *  This is responsable for removing sent commands from the transmit
 311  *  queue. This is not a required operation, but it of course helps
 312  *  reduce memory footprint at run time!
 313  *
 314  */
 315 void aim_tx_purgequeue(struct aim_session_t *sess)
 316 {
 317   struct command_tx_struct *cur = NULL;
 318   struct command_tx_struct *tmp;
 319
 320   if (sess->queue_outgoing == NULL)
 321     return;
 322
 323   if (sess->queue_outgoing->next == NULL) {
 324     if (!sess->queue_outgoing->lock && sess->queue_outgoing->sent) {
 325       tmp = sess->queue_outgoing;
 326       sess->queue_outgoing = NULL;
 327       free(tmp->data);
 328       free(tmp);
 329     }
 330     return;
 331   }
 332
 333   for(cur = sess->queue_outgoing; cur->next != NULL; ) {
 334     if (!cur->next->lock && cur->next->sent) {
 335       tmp = cur->next;
 336       cur->next = tmp->next;
 337       free(tmp->data);
 338       free(tmp);
 339     }
 340     cur = cur->next;
 341
 342     /*
 343      * Be careful here.  Because of the way we just
 344      * manipulated the pointer, cur may be NULL and
 345      * the for() will segfault doing the check unless
 346      * we find this case first.
 347      */
 348     if (cur == NULL)
 349       break;
 350   }
 351   return;
 352 }