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