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