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