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 faim_internal int aim_tx_enqueue(struct aim_session_t *sess, struct command_tx_struct *command)
 155 {
 156   /*
 157    * If we want to send a connection thats inprogress, we have to force
 158    * them to use the queue based version. Otherwise, use whatever they
 159    * want.
 160    */
 161   if (command && command->conn && (command->conn->status & AIM_CONN_STATUS_INPROGRESS)) {
 162     return aim_tx_enqueue__queuebased(sess, command);
 163   }
 164   return (*sess->tx_enqueue)(sess, command);
 165 }
 166
 167 /*
 168  *  aim_get_next_txseqnum()
 169  *
 170  *   This increments the tx command count, and returns the seqnum
 171  *   that should be stamped on the next FLAP packet sent.  This is
 172  *   normally called during the final step of packet preparation
 173  *   before enqueuement (in aim_tx_enqueue()).
 174  *
 175  */
 176 faim_internal unsigned int aim_get_next_txseqnum(struct aim_conn_t *conn)
 177 {
 178   u_int ret;
 179
 180   faim_mutex_lock(&conn->seqnum_lock);
 181   ret = ++conn->seqnum;
 182   faim_mutex_unlock(&conn->seqnum_lock);
 183   return ret;
 184 }
 185
 186 /*
 187  *  aim_tx_printqueue()
 188  *
 189  *  This is basically for debuging purposes only.  It dumps all the
 190  *  records in the tx queue and their current status.  Very helpful
 191  *  if the queue isn't working quite right.
 192  *
 193  */
 194 #if debug == 2
 195 faim_internal int aim_tx_printqueue(struct aim_session_t *sess)
 196 {
 197   struct command_tx_struct *cur;
 198
 199   faimdprintf(2, "\ncurrent aim_queue_outgoing...\n");
 200   faimdprintf(2, "\ttype seqnum  len  lock sent\n");
 201
 202   if (sess->queue_outgoing == NULL)
 203     faimdprintf(2, "aim_tx_flushqueue(): queue empty");
 204   else {
 205       for (cur = sess->queue_outgoing; cur; cur = cur->next) {
 206           faimdprintf(2, "\t  %2x  %2x   %4x %4x   %1d    %1d\n",
 207                       cur->hdrtype,
 208                       (cur->hdrtype==AIM_FRAMETYPE_OFT)?cur->hdr.oft.type:cur->hdr.oscar.type,
 209                       (cur->hdrtype==AIM_FRAMETYPE_OSCAR)?cur->seqnum:0,
 210                       cur->commandlen, cur->lock,
 211                       cur->sent);
 212       }
 213   }
 214
 215   faimdprintf(2, "\n(done printing queue)\n");
 216
 217   return 0;
 218 }
 219 #endif
 220
 221 /*
 222  *  aim_tx_flushqueue()
 223  *
 224  *  This the function is responsable for putting the queued commands
 225  *  onto the wire.  This function is critical to the operation of
 226  *  the queue and therefore is the most prone to brokenness.  It
 227  *  seems to be working quite well at this point.
 228  *
 229  *  Procedure:
 230  *    1) Traverse the list, only operate on commands that are unlocked
 231  *       and haven't been sent yet.
 232  *    2) Lock the struct
 233  *    3) Allocate a temporary buffer to store the finished, fully
 234  *       processed packet in.
 235  *    4) Build the packet from the command_tx_struct data.
 236  *    5) Write the packet to the socket.
 237  *    6) If success, mark the packet sent, if fail report failure, do NOT
 238  *       mark the packet sent (so it will not get purged and therefore
 239  *       be attempted again on next call).
 240  *    7) Unlock the struct.
 241  *    8) Free the temp buffer
 242  *    9) Step to next struct in list and go back to 1.
 243  *
 244  */
 245 faim_internal int aim_tx_sendframe(struct aim_session_t *sess, struct command_tx_struct *cur)
 246 {
 247   int buflen = 0;
 248   unsigned char *curPacket;
 249
 250   if (!cur)
 251     return -1; /* fatal */
 252
 253   cur->lock = 1; /* lock the struct */
 254
 255   if (cur->hdrtype == AIM_FRAMETYPE_OSCAR)
 256     buflen = cur->commandlen + 6;
 257   else if (cur->hdrtype == AIM_FRAMETYPE_OFT)
 258     buflen = cur->hdr.oft.hdr2len + 8;
 259   else {
 260     cur->lock = 0;
 261     return -1;
 262   }
 263
 264   /* allocate full-packet buffer */
 265   if (!(curPacket = (unsigned char *) malloc(buflen))) {
 266     cur->lock = 0;
 267     return -1;
 268   }
 269
 270   if (cur->hdrtype == AIM_FRAMETYPE_OSCAR) {
 271     /* command byte */
 272     curPacket[0] = 0x2a;
 273
 274     /* type/family byte */
 275     curPacket[1] = cur->hdr.oscar.type;
 276
 277     /* bytes 3+4: word: FLAP sequence number */
 278     aimutil_put16(curPacket+2, cur->hdr.oscar.seqnum);
 279
 280     /* bytes 5+6: word: SNAC len */
 281     aimutil_put16(curPacket+4, cur->commandlen);
 282
 283     /* bytes 7 and on: raw: SNAC data */  /* XXX: ye gods! get rid of this! */
 284     memcpy(&(curPacket[6]), cur->data, cur->commandlen);
 285
 286   } else if (cur->hdrtype == AIM_FRAMETYPE_OFT) {
 287     int z = 0;
 288
 289     z += aimutil_put8(curPacket+z, cur->hdr.oft.magic[0]);
 290     z += aimutil_put8(curPacket+z, cur->hdr.oft.magic[1]);
 291     z += aimutil_put8(curPacket+z, cur->hdr.oft.magic[2]);
 292     z += aimutil_put8(curPacket+z, cur->hdr.oft.magic[3]);
 293
 294     z += aimutil_put16(curPacket+z, cur->hdr.oft.hdr2len + 8);
 295     z += aimutil_put16(curPacket+z, cur->hdr.oft.type);
 296
 297     memcpy(curPacket+z, cur->hdr.oft.hdr2, cur->hdr.oft.hdr2len);
 298   }
 299
 300   /*
 301    * For OSCAR, a full image of the raw packet data now in curPacket.
 302    * For OFT, an image of just the bloated header is in curPacket,
 303    * since OFT allows us to do the data in a different write (yay!).
 304    */
 305   faim_mutex_lock(&cur->conn->active);
 306   if (send(cur->conn->fd, curPacket, buflen, 0) != buflen) {
 307     faim_mutex_unlock(&cur->conn->active);
 308     cur->sent = 1;
 309     aim_conn_close(cur->conn);
 310     return 0; /* bail out */
 311   }
 312
 313   if ((cur->hdrtype == AIM_FRAMETYPE_OFT) && cur->commandlen) {
 314     if (send(cur->conn->fd, cur->data, cur->commandlen, 0) != (int)cur->commandlen) {
 315       /*
 316        * Theres nothing we can do about this since we've already sent the
 317        * header!  The connection is unstable.
 318        */
 319     }
 320   }
 321
 322   cur->sent = 1; /* mark the struct as sent */
 323   cur->conn->lastactivity = time(NULL);
 324
 325   faim_mutex_unlock(&cur->conn->active);
 326
 327 #if debug > 2
 328   faimdprintf(2, "\nPacket:");
 329   for (i = 0; i < (cur->commandlen + 6); i++) {
 330     if ((i % 8) == 0) {
 331       faimdprintf(2, "\n\t");
 332     }
 333     if (curPacket[i] >= ' ' && curPacket[i]<127) {
 334       faimdprintf(2, "%c=%02x ", curPacket[i], curPacket[i]);
 335     } else {
 336       faimdprintf(2, "0x%2x ", curPacket[i]);
 337     }
 338   }
 339   faimdprintf(2, "\n");
 340 #endif
 341   cur->lock = 0; /* unlock the struct */
 342   free(curPacket); /* free up full-packet buffer */
 343
 344   return 1; /* success */
 345 }
 346
 347 faim_export int aim_tx_flushqueue(struct aim_session_t *sess)
 348 {
 349   struct command_tx_struct *cur;
 350
 351 #if debug > 1
 352   int i = 0;
 353 #endif
 354
 355   if (sess->queue_outgoing == NULL)
 356     return 0;
 357
 358   faimdprintf(2, "beginning txflush...\n");
 359   for (cur = sess->queue_outgoing; cur; cur = cur->next) {
 360     /* only process if its unlocked and unsent */
 361     if (!cur->lock && !cur->sent) {
 362
 363       if (cur->conn && (cur->conn->status & AIM_CONN_STATUS_INPROGRESS))
 364         continue;
 365
 366       /*
 367        * And now for the meager attempt to force transmit
 368        * latency and avoid missed messages.
 369        */
 370       if ((cur->conn->lastactivity + cur->conn->forcedlatency) >= time(NULL)) {
 371         /* FIXME FIXME -- should be a break! we dont want to block the upper layers */
 372         sleep((cur->conn->lastactivity + cur->conn->forcedlatency) - time(NULL));
 373       }
 374
 375       if (aim_tx_sendframe(sess, cur) == -1)
 376         break;
 377     }
 378   }
 379
 380   /* purge sent commands from queue */
 381   aim_tx_purgequeue(sess);
 382
 383   return 0;
 384 }
 385
 386 /*
 387  *  aim_tx_purgequeue()
 388  *
 389  *  This is responsable for removing sent commands from the transmit
 390  *  queue. This is not a required operation, but it of course helps
 391  *  reduce memory footprint at run time!
 392  *
 393  */
 394 faim_export void aim_tx_purgequeue(struct aim_session_t *sess)
 395 {
 396   struct command_tx_struct *cur = NULL;
 397   struct command_tx_struct *tmp;
 398
 399   if (sess->queue_outgoing == NULL)
 400     return;
 401
 402   if (sess->queue_outgoing->next == NULL) {
 403     if (!sess->queue_outgoing->lock && sess->queue_outgoing->sent) {
 404       tmp = sess->queue_outgoing;
 405       sess->queue_outgoing = NULL;
 406       if (tmp->hdrtype == AIM_FRAMETYPE_OFT)
 407         free(tmp->hdr.oft.hdr2);
 408       free(tmp->data);
 409       free(tmp);
 410     }
 411     return;
 412   }
 413
 414   for(cur = sess->queue_outgoing; cur->next != NULL; ) {
 415     if (!cur->next->lock && cur->next->sent) {
 416       tmp = cur->next;
 417       cur->next = tmp->next;
 418       if (tmp->hdrtype == AIM_FRAMETYPE_OFT)
 419         free(tmp->hdr.oft.hdr2);
 420       free(tmp->data);
 421       free(tmp);
 422     }
 423     cur = cur->next;
 424
 425     /*
 426      * Be careful here.  Because of the way we just
 427      * manipulated the pointer, cur may be NULL and
 428      * the for() will segfault doing the check unless
 429      * we find this case first.
 430      */
 431     if (cur == NULL)
 432       break;
 433   }
 434   return;
 435 }