recv_thread.c

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