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/FLAP
  24  * chan = channel for FLAP, hdrtype for OFT
  25  *
  26  */
  27 faim_internal aim_frame_t *aim_tx_new(aim_session_t *sess, aim_conn_t *conn, fu8_t framing, fu8_t chan, int datalen)
  28 {
  29         aim_frame_t *fr;
  30
  31         if (!conn) {
  32                 faimdprintf(sess, 0, "aim_tx_new: ERROR: no connection specified\n");
  33                 return NULL;
  34         }
  35
  36         /* For sanity... */
  37         if ((conn->type == AIM_CONN_TYPE_RENDEZVOUS) ||
  38                         (conn->type == AIM_CONN_TYPE_RENDEZVOUS_OUT)) {
  39                 if (framing != AIM_FRAMETYPE_OFT) {
  40                         faimdprintf(sess, 0, "aim_tx_new: attempted to allocate inappropriate frame type for rendezvous connection\n");
  41                         return NULL;
  42                 }
  43         } else {
  44                 if (framing != AIM_FRAMETYPE_FLAP) {
  45                         faimdprintf(sess, 0, "aim_tx_new: attempted to allocate inappropriate frame type for FLAP connection\n");
  46                         return NULL;
  47                 }
  48         }
  49
  50         if (!(fr = (aim_frame_t *)malloc(sizeof(aim_frame_t))))
  51                 return NULL;
  52         memset(fr, 0, sizeof(aim_frame_t));
  53
  54         fr->conn = conn;
  55
  56         fr->hdrtype = framing;
  57
  58         if (fr->hdrtype == AIM_FRAMETYPE_FLAP) {
  59
  60                 fr->hdr.flap.type = chan;
  61
  62         } else if (fr->hdrtype == AIM_FRAMETYPE_OFT) {
  63
  64                 fr->hdr.oft.type = chan;
  65                 fr->hdr.oft.hdr2len = 0; /* this will get setup by caller */
  66
  67         } else
  68                 faimdprintf(sess, 0, "tx_new: unknown framing\n");
  69
  70         if (datalen > 0) {
  71                 fu8_t *data;
  72
  73                 if (!(data = (unsigned char *)malloc(datalen))) {
  74                         aim_frame_destroy(fr);
  75                         return NULL;
  76                 }
  77
  78                 aim_bstream_init(&fr->data, data, datalen);
  79         }
  80
  81         return fr;
  82 }
  83
  84 /*
  85  * aim_tx_enqeue__queuebased()
  86  *
  87  * The overall purpose here is to enqueue the passed in command struct
  88  * into the outgoing (tx) queue.  Basically...
  89  *   1) Make a scope-irrelevent copy of the struct
  90  *   3) Mark as not-sent-yet
  91  *   4) Enqueue the struct into the list
  92  *   6) Return
  93  *
  94  * Note that this is only used when doing queue-based transmitting;
  95  * that is, when sess->tx_enqueue is set to &aim_tx_enqueue__queuebased.
  96  *
  97  */
  98 static int aim_tx_enqueue__queuebased(aim_session_t *sess, aim_frame_t *fr)
  99 {
 100
 101         if (!fr->conn) {
 102                 faimdprintf(sess, 1, "aim_tx_enqueue: WARNING: enqueueing packet with no connecetion\n");
 103                 fr->conn = aim_getconn_type(sess, AIM_CONN_TYPE_BOS);
 104         }
 105
 106         if (fr->hdrtype == AIM_FRAMETYPE_FLAP) {
 107                 /* assign seqnum -- XXX should really not assign until hardxmit */
 108                 fr->hdr.flap.seqnum = aim_get_next_txseqnum(fr->conn);
 109         }
 110
 111         fr->handled = 0; /* not sent yet */
 112
 113         /* see overhead note in aim_rxqueue counterpart */
 114         if (!sess->queue_outgoing)
 115                 sess->queue_outgoing = fr;
 116         else {
 117                 aim_frame_t *cur;
 118
 119                 for (cur = sess->queue_outgoing; cur->next; cur = cur->next)
 120                         ;
 121                 cur->next = fr;
 122         }
 123
 124         return 0;
 125 }
 126
 127 /*
 128  * aim_tx_enqueue__immediate()
 129  *
 130  * Parallel to aim_tx_enqueue__queuebased, however, this bypasses
 131  * the whole queue mess when you want immediate writes to happen.
 132  *
 133  * Basically the same as its __queuebased couterpart, however
 134  * instead of doing a list append, it just calls aim_tx_sendframe()
 135  * right here.
 136  *
 137  */
 138 static int aim_tx_enqueue__immediate(aim_session_t *sess, aim_frame_t *fr)
 139 {
 140
 141         if (!fr->conn) {
 142                 faimdprintf(sess, 1, "aim_tx_enqueue: ERROR: packet has no connection\n");
 143                 aim_frame_destroy(fr);
 144                 return 0;
 145         }
 146
 147         if (fr->hdrtype == AIM_FRAMETYPE_FLAP)
 148                 fr->hdr.flap.seqnum = aim_get_next_txseqnum(fr->conn);
 149
 150         fr->handled = 0; /* not sent yet */
 151
 152         aim_tx_sendframe(sess, fr);
 153
 154         aim_frame_destroy(fr);
 155
 156         return 0;
 157 }
 158
 159 faim_export int aim_tx_setenqueue(aim_session_t *sess, int what, int (*func)(aim_session_t *, aim_frame_t *))
 160 {
 161
 162         if (what == AIM_TX_QUEUED)
 163                 sess->tx_enqueue = &aim_tx_enqueue__queuebased;
 164         else if (what == AIM_TX_IMMEDIATE)
 165                 sess->tx_enqueue = &aim_tx_enqueue__immediate;
 166         else if (what == AIM_TX_USER) {
 167                 if (!func)
 168                         return -EINVAL;
 169                 sess->tx_enqueue = func;
 170         } else
 171                 return -EINVAL; /* unknown action */
 172
 173         return 0;
 174 }
 175
 176 faim_internal int aim_tx_enqueue(aim_session_t *sess, aim_frame_t *fr)
 177 {
 178
 179         /*
 180          * If we want to send a connection thats inprogress, we have to force
 181          * them to use the queue based version. Otherwise, use whatever they
 182          * want.
 183          */
 184         if (fr && fr->conn &&
 185                         (fr->conn->status & AIM_CONN_STATUS_INPROGRESS)) {
 186                 return aim_tx_enqueue__queuebased(sess, fr);
 187         }
 188
 189         return (*sess->tx_enqueue)(sess, fr);
 190 }
 191
 192 /*
 193  *  aim_get_next_txseqnum()
 194  *
 195  *   This increments the tx command count, and returns the seqnum
 196  *   that should be stamped on the next FLAP packet sent.  This is
 197  *   normally called during the final step of packet preparation
 198  *   before enqueuement (in aim_tx_enqueue()).
 199  *
 200  */
 201 faim_internal flap_seqnum_t aim_get_next_txseqnum(aim_conn_t *conn)
 202 {
 203         flap_seqnum_t ret;
 204
 205         faim_mutex_lock(&conn->seqnum_lock);
 206         ret = ++conn->seqnum;
 207         faim_mutex_unlock(&conn->seqnum_lock);
 208
 209         return ret;
 210 }
 211
 212 static int aim_send(int fd, const void *buf, size_t count)
 213 {
 214         int left, cur;
 215
 216         for (cur = 0, left = count; left; ) {
 217                 int ret;
 218
 219                 ret = send(fd, ((unsigned char *)buf)+cur, left, 0);
 220                 if (ret == -1)
 221                         return -1;
 222                 else if (ret == 0)
 223                         return cur;
 224
 225                 cur += ret;
 226                 left -= ret;
 227         }
 228
 229         return cur;
 230 }
 231
 232 static int aim_bstream_send(aim_bstream_t *bs, aim_conn_t *conn, size_t count)
 233 {
 234         int wrote = 0;
 235
 236         if (!bs || !conn || (count < 0))
 237                 return -EINVAL;
 238
 239         if (count > aim_bstream_empty(bs))
 240                 count = aim_bstream_empty(bs); /* truncate to remaining space */
 241
 242         if (count)
 243                 wrote = aim_send(conn->fd, bs->data + bs->offset, count);
 244
 245         if (((aim_session_t *)conn->sessv)->debug >= 2) {
 246                 int i;
 247                 aim_session_t *sess = (aim_session_t *)conn->sessv;
 248
 249                 faimdprintf(sess, 2, "\nOutgoing data: (%d bytes)", wrote);
 250                 for (i = 0; i < wrote; i++) {
 251                         if (!(i % 8))
 252                                 faimdprintf(sess, 2, "\n\t");
 253                         faimdprintf(sess, 2, "0x%02x ", *(bs->data + bs->offset + i));
 254                 }
 255                 faimdprintf(sess, 2, "\n");
 256         }
 257
 258
 259         bs->offset += wrote;
 260
 261         return wrote;
 262 }
 263
 264 static int sendframe_flap(aim_session_t *sess, aim_frame_t *fr)
 265 {
 266         aim_bstream_t obs;
 267         fu8_t *obs_raw;
 268         int payloadlen, err = 0, obslen;
 269
 270         payloadlen = aim_bstream_curpos(&fr->data);
 271
 272         if (!(obs_raw = malloc(6 + payloadlen)))
 273                 return -ENOMEM;
 274
 275         aim_bstream_init(&obs, obs_raw, 6 + payloadlen);
 276
 277         /* FLAP header */
 278         aimbs_put8(&obs, 0x2a);
 279         aimbs_put8(&obs, fr->hdr.flap.type);
 280         aimbs_put16(&obs, fr->hdr.flap.seqnum);
 281         aimbs_put16(&obs, payloadlen);
 282
 283         /* payload */
 284         aim_bstream_rewind(&fr->data);
 285         aimbs_putbs(&obs, &fr->data, payloadlen);
 286
 287         obslen = aim_bstream_curpos(&obs);
 288         aim_bstream_rewind(&obs);
 289
 290         if (aim_bstream_send(&obs, fr->conn, obslen) != obslen)
 291                 err = -errno;
 292
 293         free(obs_raw); /* XXX aim_bstream_free */
 294
 295         fr->handled = 1;
 296         fr->conn->lastactivity = time(NULL);
 297
 298         return err;
 299 }
 300
 301 static int sendframe_oft(aim_session_t *sess, aim_frame_t *fr)
 302 {
 303         aim_bstream_t hbs;
 304         fu8_t *hbs_raw;
 305         int hbslen;
 306         int err = 0;
 307
 308         hbslen = 8 + fr->hdr.oft.hdr2len;
 309
 310         if (!(hbs_raw = malloc(hbslen)))
 311                 return -1;
 312
 313         aim_bstream_init(&hbs, hbs_raw, hbslen);
 314
 315         aimbs_putraw(&hbs, fr->hdr.oft.magic, 4);
 316         aimbs_put16(&hbs, fr->hdr.oft.hdr2len + 8);
 317         aimbs_put16(&hbs, fr->hdr.oft.type);
 318         aimbs_putraw(&hbs, fr->hdr.oft.hdr2, fr->hdr.oft.hdr2len);
 319
 320         if (aim_bstream_send(&hbs, fr->conn, hbslen) != hbslen) {
 321
 322                 err = -errno;
 323
 324         } else if (aim_bstream_curpos(&fr->data)) {
 325                 int len;
 326
 327                 len = aim_bstream_curpos(&fr->data);
 328                 aim_bstream_rewind(&fr->data);
 329
 330                 if (aim_bstream_send(&fr->data, fr->conn, len) != len)
 331                         err = -errno;
 332         }
 333
 334         free(hbs_raw); /* XXX aim_bstream_free */
 335
 336         fr->handled = 1;
 337         fr->conn->lastactivity = time(NULL);
 338
 339         return err;
 340 }
 341
 342 faim_internal int aim_tx_sendframe(aim_session_t *sess, aim_frame_t *fr)
 343 {
 344         if (fr->hdrtype == AIM_FRAMETYPE_FLAP)
 345                 return sendframe_flap(sess, fr);
 346         else if (fr->hdrtype == AIM_FRAMETYPE_OFT)
 347                 return sendframe_oft(sess, fr);
 348         return -1;
 349 }
 350
 351 faim_export int aim_tx_flushqueue(aim_session_t *sess)
 352 {
 353         aim_frame_t *cur;
 354
 355         for (cur = sess->queue_outgoing; cur; cur = cur->next) {
 356
 357                 if (cur->handled)
 358                         continue; /* already been sent */
 359
 360                 if (cur->conn && (cur->conn->status & AIM_CONN_STATUS_INPROGRESS))
 361                         continue;
 362
 363                 /*
 364                  * And now for the meager attempt to force transmit
 365                  * latency and avoid missed messages.
 366                  */
 367                 if ((cur->conn->lastactivity + cur->conn->forcedlatency) >= time(NULL)) {
 368                         /*
 369                          * XXX should be a break! we dont want to block the
 370                          * upper layers
 371                          *
 372                          * XXX or better, just do this right.
 373                          *
 374                          */
 375                         sleep((cur->conn->lastactivity + cur->conn->forcedlatency) - time(NULL));
 376                 }
 377
 378                 /* XXX this should call the custom "queuing" function!! */
 379                 aim_tx_sendframe(sess, cur);
 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(aim_session_t *sess)
 397 {
 398         aim_frame_t *cur, **prev;
 399
 400         for (prev = &sess->queue_outgoing; (cur = *prev); ) {
 401
 402                 if (cur->handled) {
 403                         *prev = cur->next;
 404
 405                         aim_frame_destroy(cur);
 406
 407                 } else
 408                         prev = &cur->next;
 409         }
 410
 411         return;
 412 }
 413
 414 /**
 415  * aim_tx_cleanqueue - get rid of packets waiting for tx on a dying conn
 416  * @sess: session
 417  * @conn: connection that's dying
 418  *
 419  * for now this simply marks all packets as sent and lets them
 420  * disappear without warning.
 421  *
 422  */
 423 faim_export void aim_tx_cleanqueue(aim_session_t *sess, aim_conn_t *conn)
 424 {
 425         aim_frame_t *cur;
 426
 427         for (cur = sess->queue_outgoing; cur; cur = cur->next) {
 428                 if (cur->conn == conn)
 429                         cur->handled = 1;
 430         }
 431
 432         return;
 433 }
 434
 435