src/allocator_thread.c

   1 #undef _GNU_SOURCE
   2 #define _GNU_SOURCE
   3 #undef _POSIX_C_SOURCE
   4 #define _DARWIN_C_SOURCE
   5 #include <limits.h>
   6 #include <pthread.h>
   7 #include <unistd.h>
   8 #include <fcntl.h>
   9 #include <stdlib.h>
  10 #include <stdio.h>
  11 #include <sys/select.h>
  12 #include <assert.h>
  13 #include <string.h>
  14 #include <stdint.h>
  15 #include <stddef.h>
  16 #include <errno.h>
  17 #include <sys/mman.h>
  18 #include "allocator_thread.h"
  19 #include "debug.h"
  20 #include "ip_type.h"
  21 #include "mutex.h"
  22 #include "hash.h"
  23 #include "remotedns.h"
  24
  25 /* stuff for our internal translation table */
  26
  27 typedef struct {
  28         uint32_t hash;
  29         char* string;
  30 } string_hash_tuple;
  31
  32 typedef struct {
  33         uint32_t counter;
  34         uint32_t capa;
  35         string_hash_tuple** list;
  36 } internal_ip_lookup_table;
  37
  38 static void *dumpstring(char* s, size_t len) {
  39         char* p = malloc(len);
  40         if(p) memcpy(p, s, len);
  41         return p;
  42 }
  43
  44 static pthread_mutex_t *internal_ips_lock;
  45 static internal_ip_lookup_table *internal_ips;
  46
  47 uint32_t index_from_internal_ip(ip_type4 internalip) {
  48         PFUNC();
  49         ip_type4 tmp = internalip;
  50         uint32_t ret;
  51         ret = tmp.octet[3] + (tmp.octet[2] << 8) + (tmp.octet[1] << 16);
  52         ret -= 1;
  53         return ret;
  54 }
  55
  56 char *string_from_internal_ip(ip_type4 internalip) {
  57         PFUNC();
  58         char *res = NULL;
  59         uint32_t index = index_from_internal_ip(internalip);
  60         if(index < internal_ips->counter)
  61                 res = internal_ips->list[index]->string;
  62         return res;
  63 }
  64
  65 extern unsigned int remote_dns_subnet;
  66 ip_type4 make_internal_ip(uint32_t index) {
  67         ip_type4 ret;
  68         index++; // so we can start at .0.0.1
  69         if(index > 0xFFFFFF)
  70                 return IPT4_INVALID;
  71         ret.octet[0] = remote_dns_subnet & 0xFF;
  72         ret.octet[1] = (index & 0xFF0000) >> 16;
  73         ret.octet[2] = (index & 0xFF00) >> 8;
  74         ret.octet[3] = index & 0xFF;
  75         return ret;
  76 }
  77
  78 static ip_type4 ip_from_internal_list(char* name, size_t len) {
  79         uint32_t hash = dalias_hash((char *) name);
  80         size_t i;
  81         ip_type4 res;
  82         void* new_mem;
  83         // see if we already have this dns entry saved.
  84         if(internal_ips->counter) {
  85                 for(i = 0; i < internal_ips->counter; i++) {
  86                         if(internal_ips->list[i]->hash == hash && !strcmp(name, internal_ips->list[i]->string)) {
  87                                 res = make_internal_ip(i);
  88                                 PDEBUG("got cached ip for %s\n", name);
  89                                 goto have_ip;
  90                         }
  91                 }
  92         }
  93         // grow list if needed.
  94         if(internal_ips->capa < internal_ips->counter + 1) {
  95                 PDEBUG("realloc\n");
  96                 new_mem = realloc(internal_ips->list, (internal_ips->capa + 16) * sizeof(void *));
  97                 if(new_mem) {
  98                         internal_ips->capa += 16;
  99                         internal_ips->list = new_mem;
 100                 } else {
 101         oom:
 102                         PDEBUG("out of mem\n");
 103                         goto err_plus_unlock;
 104                 }
 105         }
 106
 107         res = make_internal_ip(internal_ips->counter);
 108         if(res.as_int == IPT4_INVALID.as_int)
 109                 goto err_plus_unlock;
 110
 111         string_hash_tuple tmp = { 0 };
 112         new_mem = dumpstring((char*) &tmp, sizeof(string_hash_tuple));
 113         if(!new_mem)
 114                 goto oom;
 115
 116         PDEBUG("creating new entry %d for ip of %s\n", (int) internal_ips->counter, name);
 117
 118         internal_ips->list[internal_ips->counter] = new_mem;
 119         internal_ips->list[internal_ips->counter]->hash = hash;
 120
 121         new_mem = dumpstring((char*) name, len);
 122
 123         if(!new_mem) {
 124                 internal_ips->list[internal_ips->counter] = 0;
 125                 goto oom;
 126         }
 127         internal_ips->list[internal_ips->counter]->string = new_mem;
 128
 129         internal_ips->counter += 1;
 130
 131         have_ip:
 132
 133         return res;
 134         err_plus_unlock:
 135
 136         PDEBUG("return err\n");
 137         return IPT4_INVALID;
 138 }
 139
 140 /* stuff for communication with the allocator thread */
 141
 142 enum at_direction {
 143         ATD_SERVER = 0,
 144         ATD_CLIENT,
 145         ATD_MAX,
 146 };
 147
 148 static pthread_t allocator_thread;
 149 int req_pipefd[2];
 150 int resp_pipefd[2];
 151
 152 static int wait_data(int readfd) {
 153         PFUNC();
 154         fd_set fds;
 155         FD_ZERO(&fds);
 156         FD_SET(readfd, &fds);
 157         int ret;
 158         while((ret = select(readfd+1, &fds, NULL, NULL, NULL)) <= 0) {
 159                 if(ret < 0) {
 160                         int e = errno;
 161                         if(e == EINTR) continue;
 162 #ifdef __GLIBC__
 163                         char emsg[1024];
 164                         char* x = strerror_r(errno, emsg, sizeof emsg);
 165                         dprintf(2, "select2: %s\n", x);
 166 #endif
 167                         return 0;
 168                 }
 169         }
 170         return 1;
 171 }
 172
 173 static int trywrite(int fd, void* buf, size_t bytes) {
 174         ssize_t ret;
 175         unsigned char *out = buf;
 176 again:
 177         ret = write(fd, out, bytes);
 178         switch(ret) {
 179                 case -1:
 180                         if(errno == EINTR) goto again;
 181                 case  0:
 182                         return 0;
 183                 default:
 184                         if(ret == bytes || !bytes) return 1;
 185                         out += ret;
 186                         bytes -= ret;
 187                         goto again;
 188         }
 189 }
 190
 191 static int sendmessage(enum at_direction dir, struct at_msg *msg) {
 192         static int* destfd[ATD_MAX] = { [ATD_SERVER] = &req_pipefd[1], [ATD_CLIENT] = &resp_pipefd[1] };
 193         assert(msg->h.datalen <= MSG_LEN_MAX);
 194         int ret = trywrite(*destfd[dir], msg, sizeof (msg->h)+msg->h.datalen);
 195         assert(msg->h.datalen <= MSG_LEN_MAX);
 196         return ret;
 197 }
 198
 199 static int tryread(int fd, void* buf, size_t bytes) {
 200         ssize_t ret;
 201         unsigned char *out = buf;
 202 again:
 203         ret = read(fd, out, bytes);
 204         switch(ret) {
 205                 case -1:
 206                         if(errno == EINTR) goto again;
 207                 case  0:
 208                         return 0;
 209                 default:
 210                         if(ret == bytes || !bytes) return 1;
 211                         out += ret;
 212                         bytes -= ret;
 213                         goto again;
 214         }
 215 }
 216 static int readmsg(int fd, struct at_msg *msg) {
 217         int ret = tryread(fd, msg, sizeof(msg->h));
 218         if(ret != 1) return ret;
 219         return tryread(fd, &msg->m, msg->h.datalen);
 220 }
 221
 222 static int getmessage(enum at_direction dir, struct at_msg *msg) {
 223         static int* readfd[ATD_MAX] = { [ATD_SERVER] = &req_pipefd[0], [ATD_CLIENT] = &resp_pipefd[0] };
 224         ssize_t ret;
 225         if((ret = wait_data(*readfd[dir]))) {
 226                 if(!readmsg(*readfd[dir], msg))
 227                         return 0;
 228                 assert(msg->h.datalen <= MSG_LEN_MAX);
 229         }
 230         return ret;
 231 }
 232
 233 static void* threadfunc(void* x) {
 234         (void) x;
 235         int ret;
 236         struct at_msg msg;
 237         while((ret = getmessage(ATD_SERVER, &msg))) {
 238                 switch(msg.h.msgtype) {
 239                         case ATM_GETIP:
 240                                 /* client wants an ip for a DNS name. iterate our list and check if we have an existing entry.
 241                                         * if not, create a new one. */
 242                                 msg.m.ip = ip_from_internal_list(msg.m.host, msg.h.datalen);
 243                                 msg.h.datalen = sizeof(ip_type4);
 244                                 break;
 245                         case ATM_GETNAME: {
 246                                 char *host = string_from_internal_ip(msg.m.ip);
 247                                 if(host) {
 248                                         size_t l = strlen(host);
 249                                         assert(l+1 < MSG_LEN_MAX);
 250                                         memcpy(msg.m.host, host, l + 1);
 251                                         msg.h.datalen = l + 1;
 252                                 } else {
 253                                         msg.h.datalen = 0;
 254                                 }
 255                                 break;
 256                         }
 257                         case ATM_EXIT:
 258                                 return 0;
 259                         default:
 260                                 abort();
 261                 }
 262                 ret = sendmessage(ATD_CLIENT, &msg);
 263         }
 264         return 0;
 265 }
 266
 267 /* API to access the internal ip mapping */
 268
 269 ip_type4 at_get_ip_for_host(char* host, size_t len) {
 270         ip_type4 readbuf;
 271         MUTEX_LOCK(internal_ips_lock);
 272         if(len > MSG_LEN_MAX) goto inv;
 273         struct at_msg msg = {.h.msgtype = ATM_GETIP, .h.datalen = len + 1 };
 274         memcpy(msg.m.host, host, len+1);
 275         if(sendmessage(ATD_SERVER, &msg) &&
 276            getmessage(ATD_CLIENT, &msg)) readbuf = msg.m.ip;
 277         else {
 278                 inv:
 279                 readbuf = IPT4_INVALID;
 280         }
 281         assert(msg.h.msgtype == ATM_GETIP);
 282         MUTEX_UNLOCK(internal_ips_lock);
 283         return readbuf;
 284 }
 285
 286 size_t at_get_host_for_ip(ip_type4 ip, char* readbuf) {
 287         struct at_msg msg = {.h.msgtype = ATM_GETNAME, .h.datalen = sizeof(ip_type4), .m.ip = ip };
 288         size_t res = 0;
 289         MUTEX_LOCK(internal_ips_lock);
 290         if(sendmessage(ATD_SERVER, &msg) && getmessage(ATD_CLIENT, &msg)) {
 291                 if((int16_t) msg.h.datalen <= 0) res = 0;
 292                 else {
 293                         memcpy(readbuf, msg.m.host, msg.h.datalen);
 294                         res = msg.h.datalen - 1;
 295                 }
 296         }
 297         assert(msg.h.msgtype == ATM_GETNAME);
 298         MUTEX_UNLOCK(internal_ips_lock);
 299         return res;
 300 }
 301
 302
 303 static void initpipe(int* fds) {
 304         int retval;
 305
 306 #ifdef HAVE_PIPE2
 307         retval = pipe2(fds, O_CLOEXEC);
 308 #else
 309         retval = pipe(fds);
 310         if(retval == 0) {
 311                 fcntl(fds[0], F_SETFD, FD_CLOEXEC);
 312                 fcntl(fds[1], F_SETFD, FD_CLOEXEC);
 313         }
 314 #endif
 315         if(retval == -1) {
 316                 perror("pipe");
 317                 exit(1);
 318         }
 319 }
 320
 321 #ifndef MAX
 322 #define MAX(x, y) ((x) > (y) ? (x) : (y))
 323 #endif
 324
 325 #if !defined(PTHREAD_STACK_MIN) || defined(__APPLE__)
 326 /* MAC says its min is 8KB, but then crashes in our face. thx hunkOLard */
 327 #define PTHREAD_STACK_MIN 64*1024
 328 #endif
 329
 330 /* initialize with pointers to shared memory. these will
 331  * be used to place responses and arguments */
 332 void at_init(void) {
 333         PFUNC();
 334         void *shm = mmap(0, 4096, PROT_WRITE|PROT_READ, MAP_ANON|MAP_SHARED, -1, 0);
 335         assert(shm);
 336         internal_ips_lock = shm;
 337         internal_ips = (void*)((char*)shm + 2048);
 338
 339         MUTEX_INIT(internal_ips_lock);
 340         memset(internal_ips, 0, sizeof *internal_ips);
 341         initpipe(req_pipefd);
 342         initpipe(resp_pipefd);
 343         pthread_attr_t allocator_thread_attr;
 344         pthread_attr_init(&allocator_thread_attr);
 345         pthread_attr_setstacksize(&allocator_thread_attr, MAX(16 * 1024, PTHREAD_STACK_MIN));
 346         pthread_create(&allocator_thread, &allocator_thread_attr, threadfunc, 0);
 347         pthread_attr_destroy(&allocator_thread_attr);
 348 }
 349
 350 void at_close(void) {
 351         PFUNC();
 352         const int msg = ATM_EXIT;
 353         write(req_pipefd[1], &msg, sizeof(int));
 354         pthread_join(allocator_thread, NULL);
 355         close(req_pipefd[0]);
 356         close(req_pipefd[1]);
 357         close(resp_pipefd[0]);
 358         close(resp_pipefd[1]);
 359         MUTEX_DESTROY(internal_ips_lock);
 360 }