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