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