dragonnet/recv_thread.c

   1 #define _GNU_SOURCE
   2 #include <assert.h>
   3 #include <endian.h>
   4 #include <pthread.h>
   5 #include <stdbool.h>
   6 #include <stdio.h>
   7 #include <stdlib.h>
   8 #include <string.h>
   9 #include <unistd.h>
  10 #include "error.h"
  11 #include "peer.h"
  12 #include "recv.h"
  13 #include "recv_thread.h"
  14 #include "sock.h"
  15
  16 void *dragonnet_peer_recv_thread(void *g_peer)
  17 {
  18 #ifdef __GLIBC__
  19         pthread_setname_np(pthread_self(), "recv");
  20 #endif
  21
  22         DragonnetPeer *p = (DragonnetPeer *) g_peer;
  23
  24         while (true) {
  25                 DragonnetTypeId type_id;
  26
  27                 bool reset = false;
  28
  29                 ssize_t len = recv(p->sock, (void *) &type_id, sizeof type_id, MSG_WAITALL);
  30                 if (len < 0) {
  31                         if (dragonnet_isconnerr()) {
  32                                 reset = true;
  33                         } else {
  34                                 dragonnet_perror("recv");
  35                                 abort();
  36                         }
  37                 }
  38
  39                 // Connection closed
  40                 if (len != sizeof type_id || reset) {
  41                         if (p->on_disconnect)
  42                                 p->on_disconnect(p);
  43
  44                         close(p->sock);
  45                         free(p->address);
  46
  47                         pthread_mutex_destroy(&p->mtx);
  48                         free(p);
  49                         return NULL;
  50                 }
  51
  52                 type_id = be16toh(type_id);
  53
  54                 if (type_id >= dragonnet_num_types) {
  55                         fprintf(stderr, "[warning] received invalid type id %d\n", type_id);
  56                         continue;
  57                 }
  58
  59                 DragonnetType type = dragonnet_types[type_id];
  60
  61                 unsigned char buf[type.siz];
  62                 memset(buf, 0, type.siz);
  63
  64                 if (!type.deserialize(p, buf)) {
  65                         if (type.free != NULL)
  66                                 type.free(buf);
  67
  68                         fprintf(stderr, "[warning] failed to deserialize package of type %d\n", type_id);
  69
  70                         continue;
  71                 }
  72
  73                 bool (*on_recv)(struct dragonnet_peer *, DragonnetTypeId, void *) = p->on_recv;
  74                 void (*on_recv_type)(DragonnetPeer *, void *) = p->on_recv_type[type_id];
  75
  76                 if (on_recv != NULL && !on_recv(p, type_id, buf))
  77                         on_recv_type = NULL;
  78
  79                 if (on_recv_type != NULL)
  80                         on_recv_type(p, buf);
  81
  82                 if (type.free != NULL)
  83                         type.free(buf);
  84         }
  85 }