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