src/network.c

   1 /**
   2  * collectd - src/network.c
   3  * Copyright (C) 2005-2007  Florian octo Forster
   4  *
   5  * This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General Public License as published by the
   7  * Free Software Foundation; only version 2 of the License is applicable.
   8  *
   9  * This program is distributed in the hope that it will be useful, but
  10  * WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU General Public License along
  15  * with this program; if not, write to the Free Software Foundation, Inc.,
  16  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  17  *
  18  * Authors:
  19  *   Florian octo Forster <octo at verplant.org>
  20  **/
  21
  22 #include "collectd.h"
  23 #include "plugin.h"
  24 #include "common.h"
  25 #include "configfile.h"
  26 #include "utils_avltree.h"
  27
  28 #include "network.h"
  29
  30 #if HAVE_PTHREAD_H
  31 # include <pthread.h>
  32 #endif
  33 #if HAVE_SYS_SOCKET_H
  34 # include <sys/socket.h>
  35 #endif
  36 #if HAVE_NETDB_H
  37 # include <netdb.h>
  38 #endif
  39 #if HAVE_NETINET_IN_H
  40 # include <netinet/in.h>
  41 #endif
  42 #if HAVE_ARPA_INET_H
  43 # include <arpa/inet.h>
  44 #endif
  45 #if HAVE_POLL_H
  46 # include <poll.h>
  47 #endif
  48
  49 /* 1500 - 40 - 8  =  Ethernet packet - IPv6 header - UDP header */
  50 /* #define BUFF_SIZE 1452 */
  51
  52 #ifndef IPV6_ADD_MEMBERSHIP
  53 # ifdef IPV6_JOIN_GROUP
  54 #  define IPV6_ADD_MEMBERSHIP IPV6_JOIN_GROUP
  55 # else
  56 #  error "Neither IP_ADD_MEMBERSHIP nor IPV6_JOIN_GROUP is defined"
  57 # endif
  58 #endif /* !IP_ADD_MEMBERSHIP */
  59
  60 #define BUFF_SIZE 1024
  61
  62 /*
  63  * Private data types
  64  */
  65 typedef struct sockent
  66 {
  67         int                      fd;
  68         struct sockaddr_storage *addr;
  69         socklen_t                addrlen;
  70         struct sockent          *next;
  71 } sockent_t;
  72
  73 /*                      1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
  74  *  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
  75  * +-------+-----------------------+-------------------------------+
  76  * ! Ver.  !                       ! Length                        !
  77  * +-------+-----------------------+-------------------------------+
  78  */
  79 struct part_header_s
  80 {
  81         uint16_t type;
  82         uint16_t length;
  83 };
  84 typedef struct part_header_s part_header_t;
  85
  86 /*                      1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
  87  *  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
  88  * +-------------------------------+-------------------------------+
  89  * ! Type                          ! Length                        !
  90  * +-------------------------------+-------------------------------+
  91  * : (Length - 4) Bytes                                            :
  92  * +---------------------------------------------------------------+
  93  */
  94 struct part_string_s
  95 {
  96         part_header_t *head;
  97         char *value;
  98 };
  99 typedef struct part_string_s part_string_t;
 100
 101 /*                      1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
 102  *  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
 103  * +-------------------------------+-------------------------------+
 104  * ! Type                          ! Length                        !
 105  * +-------------------------------+-------------------------------+
 106  * : (Length - 4 == 2 || 4 || 8) Bytes                             :
 107  * +---------------------------------------------------------------+
 108  */
 109 struct part_number_s
 110 {
 111         part_header_t *head;
 112         uint64_t *value;
 113 };
 114 typedef struct part_number_s part_number_t;
 115
 116 /*                      1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
 117  *  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
 118  * +-------------------------------+-------------------------------+
 119  * ! Type                          ! Length                        !
 120  * +-------------------------------+---------------+---------------+
 121  * ! Num of values                 ! Type0         ! Type1         !
 122  * +-------------------------------+---------------+---------------+
 123  * ! Value0                                                        !
 124  * !                                                               !
 125  * +---------------------------------------------------------------+
 126  * ! Value1                                                        !
 127  * !                                                               !
 128  * +---------------------------------------------------------------+
 129  */
 130 struct part_values_s
 131 {
 132         part_header_t *head;
 133         uint16_t *num_values;
 134         uint8_t  *values_types;
 135         value_t  *values;
 136 };
 137 typedef struct part_values_s part_values_t;
 138
 139 struct receive_list_entry_s
 140 {
 141   char data[BUFF_SIZE];
 142   int  data_len;
 143   struct receive_list_entry_s *next;
 144 };
 145 typedef struct receive_list_entry_s receive_list_entry_t;
 146
 147 /*
 148  * Private variables
 149  */
 150 static const char *config_keys[] =
 151 {
 152         "CacheFlush",
 153         "Listen",
 154         "Server",
 155         "TimeToLive",
 156         "Forward"
 157 };
 158 static int config_keys_num = STATIC_ARRAY_SIZE (config_keys);
 159
 160 static int network_config_ttl = 0;
 161 static int network_config_forward = 0;
 162
 163 static sockent_t *sending_sockets = NULL;
 164
 165 static receive_list_entry_t *receive_list_head = NULL;
 166 static receive_list_entry_t *receive_list_tail = NULL;
 167 static pthread_mutex_t       receive_list_lock = PTHREAD_MUTEX_INITIALIZER;
 168 static pthread_cond_t        receive_list_cond = PTHREAD_COND_INITIALIZER;
 169
 170 static struct pollfd *listen_sockets = NULL;
 171 static int listen_sockets_num = 0;
 172
 173 static int listen_loop = 0;
 174 static pthread_t receive_thread_id = 0;
 175 static pthread_t dispatch_thread_id = 0;
 176
 177 static char         send_buffer[BUFF_SIZE];
 178 static char        *send_buffer_ptr;
 179 static int          send_buffer_fill;
 180 static value_list_t send_buffer_vl = VALUE_LIST_STATIC;
 181 static char         send_buffer_type[DATA_MAX_NAME_LEN];
 182 static pthread_mutex_t send_buffer_lock = PTHREAD_MUTEX_INITIALIZER;
 183
 184 static c_avl_tree_t      *cache_tree = NULL;
 185 static pthread_mutex_t  cache_lock = PTHREAD_MUTEX_INITIALIZER;
 186 static time_t           cache_flush_last;
 187 static int              cache_flush_interval = 1800;
 188
 189 /*
 190  * Private functions
 191  */
 192 static int cache_flush (void)
 193 {
 194         char **keys = NULL;
 195         int    keys_num = 0;
 196
 197         char **tmp;
 198         int    i;
 199
 200         char   *key;
 201         time_t *value;
 202         c_avl_iterator_t *iter;
 203
 204         time_t curtime = time (NULL);
 205
 206         iter = c_avl_get_iterator (cache_tree);
 207         while (c_avl_iterator_next (iter, (void *) &key, (void *) &value) == 0)
 208         {
 209                 if ((curtime - *value) <= cache_flush_interval)
 210                         continue;
 211                 tmp = (char **) realloc (keys,
 212                                 (keys_num + 1) * sizeof (char *));
 213                 if (tmp == NULL)
 214                 {
 215                         sfree (keys);
 216                         c_avl_iterator_destroy (iter);
 217                         ERROR ("network plugin: cache_flush: realloc"
 218                                         " failed.");
 219                         return (-1);
 220                 }
 221                 keys = tmp;
 222                 keys[keys_num] = key;
 223                 keys_num++;
 224         } /* while (c_avl_iterator_next) */
 225         c_avl_iterator_destroy (iter);
 226
 227         for (i = 0; i < keys_num; i++)
 228         {
 229                 if (c_avl_remove (cache_tree, keys[i], (void *) &key,
 230                                         (void *) &value) != 0)
 231                 {
 232                         WARNING ("network plugin: cache_flush: c_avl_remove"
 233                                         " (%s) failed.", keys[i]);
 234                         continue;
 235                 }
 236
 237                 sfree (key);
 238                 sfree (value);
 239         }
 240
 241         sfree (keys);
 242
 243         DEBUG ("network plugin: cache_flush: Removed %i %s",
 244                         keys_num, (keys_num == 1) ? "entry" : "entries");
 245         cache_flush_last = curtime;
 246         return (0);
 247 } /* int cache_flush */
 248
 249 static int cache_check (const char *type, const value_list_t *vl)
 250 {
 251         char key[1024];
 252         time_t *value = NULL;
 253         int retval = -1;
 254
 255         if (cache_tree == NULL)
 256                 return (-1);
 257
 258         if (format_name (key, sizeof (key), vl->host, vl->plugin,
 259                                 vl->plugin_instance, type, vl->type_instance))
 260                 return (-1);
 261
 262         pthread_mutex_lock (&cache_lock);
 263
 264         if (c_avl_get (cache_tree, key, (void *) &value) == 0)
 265         {
 266                 if (*value < vl->time)
 267                 {
 268                         *value = vl->time;
 269                         retval = 0;
 270                 }
 271                 else
 272                 {
 273                         DEBUG ("network plugin: cache_check: *value = %i >= vl->time = %i",
 274                                         (int) *value, (int) vl->time);
 275                         retval = 1;
 276                 }
 277         }
 278         else
 279         {
 280                 char *key_copy = strdup (key);
 281                 value = malloc (sizeof (time_t));
 282                 if ((key_copy != NULL) && (value != NULL))
 283                 {
 284                         *value = vl->time;
 285                         c_avl_insert (cache_tree, key_copy, value);
 286                         retval = 0;
 287                 }
 288                 else
 289                 {
 290                         sfree (key_copy);
 291                         sfree (value);
 292                 }
 293         }
 294
 295         if ((time (NULL) - cache_flush_last) > cache_flush_interval)
 296                 cache_flush ();
 297
 298         pthread_mutex_unlock (&cache_lock);
 299
 300         DEBUG ("network plugin: cache_check: key = %s; time = %i; retval = %i",
 301                         key, (int) vl->time, retval);
 302
 303         return (retval);
 304 } /* int cache_check */
 305
 306 static int write_part_values (char **ret_buffer, int *ret_buffer_len,
 307                 const data_set_t *ds, const value_list_t *vl)
 308 {
 309         char *packet_ptr;
 310         int packet_len;
 311         int num_values;
 312
 313         part_header_t pkg_ph;
 314         uint16_t      pkg_num_values;
 315         uint8_t      *pkg_values_types;
 316         value_t      *pkg_values;
 317
 318         int offset;
 319         int i;
 320
 321         num_values = vl->values_len;
 322         packet_len = sizeof (part_header_t) + sizeof (uint16_t)
 323                 + (num_values * sizeof (uint8_t))
 324                 + (num_values * sizeof (value_t));
 325
 326         if (*ret_buffer_len < packet_len)
 327                 return (-1);
 328
 329         pkg_values_types = (uint8_t *) malloc (num_values * sizeof (uint8_t));
 330         if (pkg_values_types == NULL)
 331         {
 332                 ERROR ("network plugin: write_part_values: malloc failed.");
 333                 return (-1);
 334         }
 335
 336         pkg_values = (value_t *) malloc (num_values * sizeof (value_t));
 337         if (pkg_values == NULL)
 338         {
 339                 free (pkg_values_types);
 340                 ERROR ("network plugin: write_part_values: malloc failed.");
 341                 return (-1);
 342         }
 343
 344         pkg_ph.type = htons (TYPE_VALUES);
 345         pkg_ph.length = htons (packet_len);
 346
 347         pkg_num_values = htons ((uint16_t) vl->values_len);
 348
 349         for (i = 0; i < num_values; i++)
 350         {
 351                 if (ds->ds[i].type == DS_TYPE_COUNTER)
 352                 {
 353                         pkg_values_types[i] = DS_TYPE_COUNTER;
 354                         pkg_values[i].counter = htonll (vl->values[i].counter);
 355                 }
 356                 else
 357                 {
 358                         pkg_values_types[i] = DS_TYPE_GAUGE;
 359                         pkg_values[i].gauge = vl->values[i].gauge;
 360                 }
 361         }
 362
 363         /*
 364          * Use `memcpy' to write everything to the buffer, because the pointer
 365          * may be unaligned and some architectures, such as SPARC, can't handle
 366          * that.
 367          */
 368         packet_ptr = *ret_buffer;
 369         offset = 0;
 370         memcpy (packet_ptr + offset, &pkg_ph, sizeof (pkg_ph));
 371         offset += sizeof (pkg_ph);
 372         memcpy (packet_ptr + offset, &pkg_num_values, sizeof (pkg_num_values));
 373         offset += sizeof (pkg_num_values);
 374         memcpy (packet_ptr + offset, pkg_values_types, num_values * sizeof (uint8_t));
 375         offset += num_values * sizeof (uint8_t);
 376         memcpy (packet_ptr + offset, pkg_values, num_values * sizeof (value_t));
 377         offset += num_values * sizeof (value_t);
 378
 379         assert (offset == packet_len);
 380
 381         *ret_buffer = packet_ptr + packet_len;
 382         *ret_buffer_len -= packet_len;
 383
 384         free (pkg_values_types);
 385         free (pkg_values);
 386
 387         return (0);
 388 } /* int write_part_values */
 389
 390 static int write_part_number (char **ret_buffer, int *ret_buffer_len,
 391                 int type, uint64_t value)
 392 {
 393         char *packet_ptr;
 394         int packet_len;
 395
 396         part_header_t pkg_head;
 397         uint64_t pkg_value;
 398
 399         int offset;
 400
 401         packet_len = sizeof (pkg_head) + sizeof (pkg_value);
 402
 403         if (*ret_buffer_len < packet_len)
 404                 return (-1);
 405
 406         pkg_head.type = htons (type);
 407         pkg_head.length = htons (packet_len);
 408         pkg_value = htonll (value);
 409
 410         packet_ptr = *ret_buffer;
 411         offset = 0;
 412         memcpy (packet_ptr + offset, &pkg_head, sizeof (pkg_head));
 413         offset += sizeof (pkg_head);
 414         memcpy (packet_ptr + offset, &pkg_value, sizeof (pkg_value));
 415         offset += sizeof (pkg_value);
 416
 417         assert (offset == packet_len);
 418
 419         *ret_buffer = packet_ptr + packet_len;
 420         *ret_buffer_len -= packet_len;
 421
 422         return (0);
 423 } /* int write_part_number */
 424
 425 static int write_part_string (char **ret_buffer, int *ret_buffer_len,
 426                 int type, const char *str, int str_len)
 427 {
 428         char *packet_ptr;
 429         int packet_len;
 430
 431         part_header_t pkg_head;
 432
 433         int offset;
 434
 435         packet_len = sizeof (pkg_head) + str_len + 1;
 436         if (*ret_buffer_len < packet_len)
 437                 return (-1);
 438
 439         pkg_head.type = htons (type);
 440         pkg_head.length = htons (packet_len);
 441
 442         packet_ptr = *ret_buffer;
 443         offset = 0;
 444         memcpy (packet_ptr + offset, &pkg_head, sizeof (pkg_head));
 445         offset += sizeof (pkg_head);
 446         memcpy (packet_ptr + offset, str, str_len);
 447         offset += str_len;
 448         memset (packet_ptr + offset, '\0', 1);
 449         offset += 1;
 450
 451         assert (offset == packet_len);
 452
 453         *ret_buffer = packet_ptr + packet_len;
 454         *ret_buffer_len -= packet_len;
 455
 456         return (0);
 457 } /* int write_part_string */
 458
 459 static int parse_part_values (void **ret_buffer, int *ret_buffer_len,
 460                 value_t **ret_values, int *ret_num_values)
 461 {
 462         char *buffer = *ret_buffer;
 463         int   buffer_len = *ret_buffer_len;
 464         part_values_t pv;
 465         int   i;
 466
 467         uint16_t h_length;
 468         uint16_t h_type;
 469         uint16_t h_num;
 470
 471         if (buffer_len < (15))
 472         {
 473                 DEBUG ("network plugin: packet is too short: buffer_len = %i",
 474                                 buffer_len);
 475                 return (-1);
 476         }
 477
 478         pv.head = (part_header_t *) buffer;
 479         h_length = ntohs (pv.head->length);
 480         h_type = ntohs (pv.head->type);
 481
 482         assert (h_type == TYPE_VALUES);
 483
 484         pv.num_values = (uint16_t *) (pv.head + 1);
 485         h_num = ntohs (*pv.num_values);
 486
 487         if (h_num != ((h_length - 6) / 9))
 488         {
 489                 DEBUG ("`length' and `num of values' don't match");
 490                 return (-1);
 491         }
 492
 493         pv.values_types = (uint8_t *) (pv.num_values + 1);
 494         pv.values = (value_t *) (pv.values_types + h_num);
 495
 496         for (i = 0; i < h_num; i++)
 497                 if (pv.values_types[i] == DS_TYPE_COUNTER)
 498                         pv.values[i].counter = ntohll (pv.values[i].counter);
 499
 500         *ret_buffer     = (void *) (pv.values + h_num);
 501         *ret_buffer_len = buffer_len - h_length;
 502         *ret_num_values = h_num;
 503         *ret_values     = pv.values;
 504
 505         return (0);
 506 } /* int parse_part_values */
 507
 508 static int parse_part_number (void **ret_buffer, int *ret_buffer_len,
 509                 uint64_t *value)
 510 {
 511         part_number_t pn;
 512         uint16_t len;
 513
 514         pn.head = (part_header_t *) *ret_buffer;
 515         pn.value = (uint64_t *) (pn.head + 1);
 516
 517         len = ntohs (pn.head->length);
 518         if (len != 12)
 519                 return (-1);
 520         if (len > *ret_buffer_len)
 521                 return (-1);
 522         *value = ntohll (*pn.value);
 523
 524         *ret_buffer = (void *) (pn.value + 1);
 525         *ret_buffer_len -= len;
 526
 527         return (0);
 528 } /* int parse_part_number */
 529
 530 static int parse_part_string (void **ret_buffer, int *ret_buffer_len,
 531                 char *output, int output_len)
 532 {
 533         char *buffer = *ret_buffer;
 534         int   buffer_len = *ret_buffer_len;
 535         part_string_t ps;
 536
 537         uint16_t h_length;
 538         uint16_t h_type;
 539
 540         DEBUG ("network plugin: parse_part_string: ret_buffer = %p;"
 541                         " ret_buffer_len = %i; output = %p; output_len = %i;",
 542                         *ret_buffer, *ret_buffer_len,
 543                         (void *) output, output_len);
 544
 545         ps.head = (part_header_t *) buffer;
 546
 547         h_length = ntohs (ps.head->length);
 548         h_type = ntohs (ps.head->type);
 549
 550         DEBUG ("network plugin: parse_part_string: length = %hu; type = %hu;",
 551                         h_length, h_type);
 552
 553         if (buffer_len < h_length)
 554         {
 555                 DEBUG ("packet is too short");
 556                 return (-1);
 557         }
 558         assert ((h_type == TYPE_HOST)
 559                         || (h_type == TYPE_PLUGIN)
 560                         || (h_type == TYPE_PLUGIN_INSTANCE)
 561                         || (h_type == TYPE_TYPE)
 562                         || (h_type == TYPE_TYPE_INSTANCE));
 563
 564         ps.value = buffer + 4;
 565         if (ps.value[h_length - 5] != '\0')
 566         {
 567                 DEBUG ("String does not end with a nullbyte");
 568                 return (-1);
 569         }
 570
 571         if (output_len < (h_length - 4))
 572         {
 573                 DEBUG ("output buffer is too small");
 574                 return (-1);
 575         }
 576         strcpy (output, ps.value);
 577
 578         DEBUG ("network plugin: parse_part_string: output = %s", output);
 579
 580         *ret_buffer = (void *) (buffer + h_length);
 581         *ret_buffer_len = buffer_len - h_length;
 582
 583         return (0);
 584 } /* int parse_part_string */
 585
 586 static int parse_packet (void *buffer, int buffer_len)
 587 {
 588         part_header_t *header;
 589         int status;
 590
 591         value_list_t vl = VALUE_LIST_INIT;
 592         char type[DATA_MAX_NAME_LEN];
 593
 594         DEBUG ("network plugin: parse_packet: buffer = %p; buffer_len = %i;",
 595                         buffer, buffer_len);
 596
 597         memset (&vl, '\0', sizeof (vl));
 598         memset (&type, '\0', sizeof (type));
 599         status = 0;
 600
 601         while ((status == 0) && (buffer_len > sizeof (part_header_t)))
 602         {
 603                 header = (part_header_t *) buffer;
 604
 605                 if (ntohs (header->length) > buffer_len)
 606                         break;
 607                 /* Assure that this loop terminates eventually */
 608                 if (ntohs (header->length) < 4)
 609                         break;
 610
 611                 if (ntohs (header->type) == TYPE_VALUES)
 612                 {
 613                         status = parse_part_values (&buffer, &buffer_len,
 614                                         &vl.values, &vl.values_len);
 615
 616                         if (status != 0)
 617                         {
 618                                 DEBUG ("parse_part_values failed.");
 619                                 break;
 620                         }
 621
 622                         if ((vl.time > 0)
 623                                         && (strlen (vl.host) > 0)
 624                                         && (strlen (vl.plugin) > 0)
 625                                         && (strlen (type) > 0)
 626                                         && (cache_check (type, &vl) == 0))
 627                         {
 628                                 DEBUG ("network plugin: parse_packet:"
 629                                                 " dispatching values");
 630                                 plugin_dispatch_values (type, &vl);
 631                         }
 632                         else
 633                         {
 634                                 DEBUG ("network plugin: parse_packet:"
 635                                                 " NOT dispatching values");
 636                         }
 637                 }
 638                 else if (ntohs (header->type) == TYPE_TIME)
 639                 {
 640                         uint64_t tmp = 0;
 641                         status = parse_part_number (&buffer, &buffer_len, &tmp);
 642                         if (status == 0)
 643                                 vl.time = (time_t) tmp;
 644                 }
 645                 else if (ntohs (header->type) == TYPE_INTERVAL)
 646                 {
 647                         uint64_t tmp = 0;
 648                         status = parse_part_number (&buffer, &buffer_len, &tmp);
 649                         if (status == 0)
 650                                 vl.interval = (int) tmp;
 651                 }
 652                 else if (ntohs (header->type) == TYPE_HOST)
 653                 {
 654                         status = parse_part_string (&buffer, &buffer_len,
 655                                         vl.host, sizeof (vl.host));
 656                         DEBUG ("network plugin: parse_packet: vl.host = %s", vl.host);
 657                 }
 658                 else if (ntohs (header->type) == TYPE_PLUGIN)
 659                 {
 660                         status = parse_part_string (&buffer, &buffer_len,
 661                                         vl.plugin, sizeof (vl.plugin));
 662                         DEBUG ("network plugin: parse_packet: vl.plugin = %s", vl.plugin);
 663                 }
 664                 else if (ntohs (header->type) == TYPE_PLUGIN_INSTANCE)
 665                 {
 666                         status = parse_part_string (&buffer, &buffer_len,
 667                                         vl.plugin_instance, sizeof (vl.plugin_instance));
 668                         DEBUG ("network plugin: parse_packet: vl.plugin_instance = %s", vl.plugin_instance);
 669                 }
 670                 else if (ntohs (header->type) == TYPE_TYPE)
 671                 {
 672                         status = parse_part_string (&buffer, &buffer_len,
 673                                         type, sizeof (type));
 674                         DEBUG ("network plugin: parse_packet: type = %s", type);
 675                 }
 676                 else if (ntohs (header->type) == TYPE_TYPE_INSTANCE)
 677                 {
 678                         status = parse_part_string (&buffer, &buffer_len,
 679                                         vl.type_instance, sizeof (vl.type_instance));
 680                         DEBUG ("network plugin: parse_packet: vl.type_instance = %s", vl.type_instance);
 681                 }
 682                 else
 683                 {
 684                         DEBUG ("network plugin: parse_packet: Unknown part"
 685                                         " type: 0x%0hx", ntohs (header->type));
 686                         buffer = ((char *) buffer) + ntohs (header->length);
 687                 }
 688         } /* while (buffer_len > sizeof (part_header_t)) */
 689
 690         return (0);
 691 } /* int parse_packet */
 692
 693 static void free_sockent (sockent_t *se)
 694 {
 695         sockent_t *next;
 696         while (se != NULL)
 697         {
 698                 next = se->next;
 699                 free (se->addr);
 700                 free (se);
 701                 se = next;
 702         }
 703 } /* void free_sockent */
 704
 705 /*
 706  * int network_set_ttl
 707  *
 708  * Set the `IP_MULTICAST_TTL', `IP_TTL', `IPV6_MULTICAST_HOPS' or
 709  * `IPV6_UNICAST_HOPS', depending on which option is applicable.
 710  *
 711  * The `struct addrinfo' is used to destinguish between unicast and multicast
 712  * sockets.
 713  */
 714 static int network_set_ttl (const sockent_t *se, const struct addrinfo *ai)
 715 {
 716         if ((network_config_ttl < 1) || (network_config_ttl > 255))
 717                 return (-1);
 718
 719         DEBUG ("ttl = %i", network_config_ttl);
 720
 721         if (ai->ai_family == AF_INET)
 722         {
 723                 struct sockaddr_in *addr = (struct sockaddr_in *) ai->ai_addr;
 724                 int optname;
 725
 726                 if (IN_MULTICAST (ntohl (addr->sin_addr.s_addr)))
 727                         optname = IP_MULTICAST_TTL;
 728                 else
 729                         optname = IP_TTL;
 730
 731                 if (setsockopt (se->fd, IPPROTO_IP, optname,
 732                                         &network_config_ttl,
 733                                         sizeof (network_config_ttl)) == -1)
 734                 {
 735                         char errbuf[1024];
 736                         ERROR ("setsockopt: %s",
 737                                         sstrerror (errno, errbuf, sizeof (errbuf)));
 738                         return (-1);
 739                 }
 740         }
 741         else if (ai->ai_family == AF_INET6)
 742         {
 743                 /* Useful example: http://gsyc.escet.urjc.es/~eva/IPv6-web/examples/mcast.html */
 744                 struct sockaddr_in6 *addr = (struct sockaddr_in6 *) ai->ai_addr;
 745                 int optname;
 746
 747                 if (IN6_IS_ADDR_MULTICAST (&addr->sin6_addr))
 748                         optname = IPV6_MULTICAST_HOPS;
 749                 else
 750                         optname = IPV6_UNICAST_HOPS;
 751
 752                 if (setsockopt (se->fd, IPPROTO_IPV6, optname,
 753                                         &network_config_ttl,
 754                                         sizeof (network_config_ttl)) == -1)
 755                 {
 756                         char errbuf[1024];
 757                         ERROR ("setsockopt: %s",
 758                                         sstrerror (errno, errbuf,
 759                                                 sizeof (errbuf)));
 760                         return (-1);
 761                 }
 762         }
 763
 764         return (0);
 765 } /* int network_set_ttl */
 766
 767 static int network_bind_socket (const sockent_t *se, const struct addrinfo *ai)
 768 {
 769         int loop = 0;
 770
 771         DEBUG ("fd = %i; calling `bind'", se->fd);
 772
 773         if (bind (se->fd, ai->ai_addr, ai->ai_addrlen) == -1)
 774         {
 775                 char errbuf[1024];
 776                 ERROR ("bind: %s",
 777                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 778                 return (-1);
 779         }
 780
 781         if (ai->ai_family == AF_INET)
 782         {
 783                 struct sockaddr_in *addr = (struct sockaddr_in *) ai->ai_addr;
 784                 if (IN_MULTICAST (ntohl (addr->sin_addr.s_addr)))
 785                 {
 786                         struct ip_mreq mreq;
 787
 788                         DEBUG ("fd = %i; IPv4 multicast address found", se->fd);
 789
 790                         mreq.imr_multiaddr.s_addr = addr->sin_addr.s_addr;
 791                         mreq.imr_interface.s_addr = htonl (INADDR_ANY);
 792
 793                         if (setsockopt (se->fd, IPPROTO_IP, IP_MULTICAST_LOOP,
 794                                                 &loop, sizeof (loop)) == -1)
 795                         {
 796                                 char errbuf[1024];
 797                                 ERROR ("setsockopt: %s",
 798                                                 sstrerror (errno, errbuf,
 799                                                         sizeof (errbuf)));
 800                                 return (-1);
 801                         }
 802
 803                         if (setsockopt (se->fd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
 804                                                 &mreq, sizeof (mreq)) == -1)
 805                         {
 806                                 char errbuf[1024];
 807                                 ERROR ("setsockopt: %s",
 808                                                 sstrerror (errno, errbuf,
 809                                                         sizeof (errbuf)));
 810                                 return (-1);
 811                         }
 812                 }
 813         }
 814         else if (ai->ai_family == AF_INET6)
 815         {
 816                 /* Useful example: http://gsyc.escet.urjc.es/~eva/IPv6-web/examples/mcast.html */
 817                 struct sockaddr_in6 *addr = (struct sockaddr_in6 *) ai->ai_addr;
 818                 if (IN6_IS_ADDR_MULTICAST (&addr->sin6_addr))
 819                 {
 820                         struct ipv6_mreq mreq;
 821
 822                         DEBUG ("fd = %i; IPv6 multicast address found", se->fd);
 823
 824                         memcpy (&mreq.ipv6mr_multiaddr,
 825                                         &addr->sin6_addr,
 826                                         sizeof (addr->sin6_addr));
 827
 828                         /* http://developer.apple.com/documentation/Darwin/Reference/ManPages/man4/ip6.4.html
 829                          * ipv6mr_interface may be set to zeroes to
 830                          * choose the default multicast interface or to
 831                          * the index of a particular multicast-capable
 832                          * interface if the host is multihomed.
 833                          * Membership is associ-associated with a
 834                          * single interface; programs running on
 835                          * multihomed hosts may need to join the same
 836                          * group on more than one interface.*/
 837                         mreq.ipv6mr_interface = 0;
 838
 839                         if (setsockopt (se->fd, IPPROTO_IPV6, IPV6_MULTICAST_LOOP,
 840                                                 &loop, sizeof (loop)) == -1)
 841                         {
 842                                 char errbuf[1024];
 843                                 ERROR ("setsockopt: %s",
 844                                                 sstrerror (errno, errbuf,
 845                                                         sizeof (errbuf)));
 846                                 return (-1);
 847                         }
 848
 849                         if (setsockopt (se->fd, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP,
 850                                                 &mreq, sizeof (mreq)) == -1)
 851                         {
 852                                 char errbuf[1024];
 853                                 ERROR ("setsockopt: %s",
 854                                                 sstrerror (errno, errbuf,
 855                                                         sizeof (errbuf)));
 856                                 return (-1);
 857                         }
 858                 }
 859         }
 860
 861         return (0);
 862 } /* int network_bind_socket */
 863
 864 static sockent_t *network_create_socket (const char *node,
 865                 const char *service,
 866                 int listen)
 867 {
 868         struct addrinfo  ai_hints;
 869         struct addrinfo *ai_list, *ai_ptr;
 870         int              ai_return;
 871
 872         sockent_t *se_head = NULL;
 873         sockent_t *se_tail = NULL;
 874
 875         DEBUG ("node = %s, service = %s", node, service);
 876
 877         memset (&ai_hints, '\0', sizeof (ai_hints));
 878         ai_hints.ai_flags    = 0;
 879 #ifdef AI_PASSIVE
 880         ai_hints.ai_flags |= AI_PASSIVE;
 881 #endif
 882 #ifdef AI_ADDRCONFIG
 883         ai_hints.ai_flags |= AI_ADDRCONFIG;
 884 #endif
 885         ai_hints.ai_family   = AF_UNSPEC;
 886         ai_hints.ai_socktype = SOCK_DGRAM;
 887         ai_hints.ai_protocol = IPPROTO_UDP;
 888
 889         ai_return = getaddrinfo (node, service, &ai_hints, &ai_list);
 890         if (ai_return != 0)
 891         {
 892                 char errbuf[1024];
 893                 ERROR ("getaddrinfo (%s, %s): %s",
 894                                 (node == NULL) ? "(null)" : node,
 895                                 (service == NULL) ? "(null)" : service,
 896                                 (ai_return == EAI_SYSTEM)
 897                                 ? sstrerror (errno, errbuf, sizeof (errbuf))
 898                                 : gai_strerror (ai_return));
 899                 return (NULL);
 900         }
 901
 902         for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
 903         {
 904                 sockent_t *se;
 905
 906                 if ((se = (sockent_t *) malloc (sizeof (sockent_t))) == NULL)
 907                 {
 908                         char errbuf[1024];
 909                         ERROR ("malloc: %s",
 910                                         sstrerror (errno, errbuf,
 911                                                 sizeof (errbuf)));
 912                         continue;
 913                 }
 914
 915                 if ((se->addr = (struct sockaddr_storage *) malloc (sizeof (struct sockaddr_storage))) == NULL)
 916                 {
 917                         char errbuf[1024];
 918                         ERROR ("malloc: %s",
 919                                         sstrerror (errno, errbuf,
 920                                                 sizeof (errbuf)));
 921                         free (se);
 922                         continue;
 923                 }
 924
 925                 assert (sizeof (struct sockaddr_storage) >= ai_ptr->ai_addrlen);
 926                 memset (se->addr, '\0', sizeof (struct sockaddr_storage));
 927                 memcpy (se->addr, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
 928                 se->addrlen = ai_ptr->ai_addrlen;
 929
 930                 se->fd   = socket (ai_ptr->ai_family,
 931                                 ai_ptr->ai_socktype,
 932                                 ai_ptr->ai_protocol);
 933                 se->next = NULL;
 934
 935                 if (se->fd == -1)
 936                 {
 937                         char errbuf[1024];
 938                         ERROR ("socket: %s",
 939                                         sstrerror (errno, errbuf,
 940                                                 sizeof (errbuf)));
 941                         free (se->addr);
 942                         free (se);
 943                         continue;
 944                 }
 945
 946                 if (listen != 0)
 947                 {
 948                         if (network_bind_socket (se, ai_ptr) != 0)
 949                         {
 950                                 close (se->fd);
 951                                 free (se->addr);
 952                                 free (se);
 953                                 continue;
 954                         }
 955                 }
 956                 else /* listen == 0 */
 957                 {
 958                         network_set_ttl (se, ai_ptr);
 959                 }
 960
 961                 if (se_tail == NULL)
 962                 {
 963                         se_head = se;
 964                         se_tail = se;
 965                 }
 966                 else
 967                 {
 968                         se_tail->next = se;
 969                         se_tail = se;
 970                 }
 971
 972                 /* We don't open more than one write-socket per node/service pair.. */
 973                 if (listen == 0)
 974                         break;
 975         }
 976
 977         freeaddrinfo (ai_list);
 978
 979         return (se_head);
 980 } /* sockent_t *network_create_socket */
 981
 982 static sockent_t *network_create_default_socket (int listen)
 983 {
 984         sockent_t *se_ptr  = NULL;
 985         sockent_t *se_head = NULL;
 986         sockent_t *se_tail = NULL;
 987
 988         se_ptr = network_create_socket (NET_DEFAULT_V6_ADDR,
 989                         NET_DEFAULT_PORT, listen);
 990
 991         /* Don't send to the same machine in IPv6 and IPv4 if both are available. */
 992         if ((listen == 0) && (se_ptr != NULL))
 993                 return (se_ptr);
 994
 995         if (se_ptr != NULL)
 996         {
 997                 se_head = se_ptr;
 998                 se_tail = se_ptr;
 999                 while (se_tail->next != NULL)
1000                         se_tail = se_tail->next;
1001         }
1002
1003         se_ptr = network_create_socket (NET_DEFAULT_V4_ADDR, NET_DEFAULT_PORT, listen);
1004
1005         if (se_tail == NULL)
1006                 return (se_ptr);
1007
1008         se_tail->next = se_ptr;
1009         return (se_head);
1010 } /* sockent_t *network_create_default_socket */
1011
1012 static int network_add_listen_socket (const char *node, const char *service)
1013 {
1014         sockent_t *se;
1015         sockent_t *se_ptr;
1016         int se_num = 0;
1017
1018         if (service == NULL)
1019                 service = NET_DEFAULT_PORT;
1020
1021         if (node == NULL)
1022                 se = network_create_default_socket (1 /* listen == true */);
1023         else
1024                 se = network_create_socket (node, service, 1 /* listen == true */);
1025
1026         if (se == NULL)
1027                 return (-1);
1028
1029         for (se_ptr = se; se_ptr != NULL; se_ptr = se_ptr->next)
1030                 se_num++;
1031
1032         listen_sockets = (struct pollfd *) realloc (listen_sockets,
1033                         (listen_sockets_num + se_num)
1034                         * sizeof (struct pollfd));
1035
1036         for (se_ptr = se; se_ptr != NULL; se_ptr = se_ptr->next)
1037         {
1038                 listen_sockets[listen_sockets_num].fd = se_ptr->fd;
1039                 listen_sockets[listen_sockets_num].events = POLLIN | POLLPRI;
1040                 listen_sockets[listen_sockets_num].revents = 0;
1041                 listen_sockets_num++;
1042         } /* for (se) */
1043
1044         free_sockent (se);
1045         return (0);
1046 } /* int network_add_listen_socket */
1047
1048 static int network_add_sending_socket (const char *node, const char *service)
1049 {
1050         sockent_t *se;
1051         sockent_t *se_ptr;
1052
1053         if (service == NULL)
1054                 service = NET_DEFAULT_PORT;
1055
1056         if (node == NULL)
1057                 se = network_create_default_socket (0 /* listen == false */);
1058         else
1059                 se = network_create_socket (node, service, 0 /* listen == false */);
1060
1061         if (se == NULL)
1062                 return (-1);
1063
1064         if (sending_sockets == NULL)
1065         {
1066                 sending_sockets = se;
1067                 return (0);
1068         }
1069
1070         for (se_ptr = sending_sockets; se_ptr->next != NULL; se_ptr = se_ptr->next)
1071                 /* seek end */;
1072
1073         se_ptr->next = se;
1074         return (0);
1075 } /* int network_get_listen_socket */
1076
1077 static void *dispatch_thread (void *arg)
1078 {
1079   while (42)
1080   {
1081     receive_list_entry_t *ent;
1082
1083     /* Lock and wait for more data to come in */
1084     pthread_mutex_lock (&receive_list_lock);
1085     while ((listen_loop == 0)
1086         && (receive_list_head == NULL))
1087       pthread_cond_wait (&receive_list_cond, &receive_list_lock);
1088
1089     /* Remove the head entry and unlock */
1090     ent = receive_list_head;
1091     if (ent != NULL)
1092       receive_list_head = ent->next;
1093     pthread_mutex_unlock (&receive_list_lock);
1094
1095     /* Check whether we are supposed to exit. We do NOT check `listen_loop'
1096      * because we dispatch all missing packets before shutting down. */
1097     if (ent == NULL)
1098       break;
1099
1100     parse_packet (ent->data, ent->data_len);
1101
1102     sfree (ent);
1103   } /* while (42) */
1104
1105   return (NULL);
1106 } /* void *receive_thread */
1107
1108 static int network_receive (void)
1109 {
1110         char buffer[BUFF_SIZE];
1111         int  buffer_len;
1112
1113         int i;
1114         int status;
1115
1116         if (listen_sockets_num == 0)
1117                 network_add_listen_socket (NULL, NULL);
1118
1119         if (listen_sockets_num == 0)
1120         {
1121                 ERROR ("network: Failed to open a listening socket.");
1122                 return (-1);
1123         }
1124
1125         while (listen_loop == 0)
1126         {
1127                 status = poll (listen_sockets, listen_sockets_num, -1);
1128
1129                 if (status <= 0)
1130                 {
1131                         char errbuf[1024];
1132                         if (errno == EINTR)
1133                                 continue;
1134                         ERROR ("poll failed: %s",
1135                                         sstrerror (errno, errbuf, sizeof (errbuf)));
1136                         return (-1);
1137                 }
1138
1139                 for (i = 0; (i < listen_sockets_num) && (status > 0); i++)
1140                 {
1141                         receive_list_entry_t *ent;
1142
1143                         if ((listen_sockets[i].revents & (POLLIN | POLLPRI)) == 0)
1144                                 continue;
1145                         status--;
1146
1147                         buffer_len = recv (listen_sockets[i].fd,
1148                                         buffer, sizeof (buffer),
1149                                         0 /* no flags */);
1150                         if (buffer_len < 0)
1151                         {
1152                                 char errbuf[1024];
1153                                 ERROR ("recv failed: %s",
1154                                                 sstrerror (errno, errbuf,
1155                                                         sizeof (errbuf)));
1156                                 return (-1);
1157                         }
1158
1159                         ent = malloc (sizeof (receive_list_entry_t));
1160                         if (ent == NULL)
1161                         {
1162                                 ERROR ("network plugin: malloc failed.");
1163                                 return (-1);
1164                         }
1165                         memset (ent, '\0', sizeof (receive_list_entry_t));
1166
1167                         /* Hopefully this be optimized out by the compiler. It
1168                          * might help prevent stupid bugs in the future though.
1169                          */
1170                         assert (sizeof (ent->data) == sizeof (buffer));
1171
1172                         memcpy (ent->data, buffer, buffer_len);
1173                         ent->data_len = buffer_len;
1174
1175                         pthread_mutex_lock (&receive_list_lock);
1176                         if (receive_list_head == NULL)
1177                         {
1178                                 receive_list_head = ent;
1179                                 receive_list_tail = ent;
1180                         }
1181                         else
1182                         {
1183                                 receive_list_tail->next = ent;
1184                                 receive_list_tail = ent;
1185                         }
1186                         pthread_cond_signal (&receive_list_cond);
1187                         pthread_mutex_unlock (&receive_list_lock);
1188                 } /* for (listen_sockets) */
1189         } /* while (listen_loop == 0) */
1190
1191         return (0);
1192 }
1193
1194 static void *receive_thread (void *arg)
1195 {
1196         return (network_receive () ? (void *) 1 : (void *) 0);
1197 } /* void *receive_thread */
1198
1199 static void network_send_buffer (const char *buffer, int buffer_len)
1200 {
1201         sockent_t *se;
1202         int status;
1203
1204         DEBUG ("network plugin: network_send_buffer: buffer_len = %i", buffer_len);
1205
1206         for (se = sending_sockets; se != NULL; se = se->next)
1207         {
1208                 while (42)
1209                 {
1210                         status = sendto (se->fd, buffer, buffer_len, 0 /* no flags */,
1211                                         (struct sockaddr *) se->addr, se->addrlen);
1212                         if (status < 0)
1213                         {
1214                                 char errbuf[1024];
1215                                 if (errno == EINTR)
1216                                         continue;
1217                                 ERROR ("network plugin: sendto failed: %s",
1218                                                 sstrerror (errno, errbuf,
1219                                                         sizeof (errbuf)));
1220                                 break;
1221                         }
1222
1223                         break;
1224                 } /* while (42) */
1225         } /* for (sending_sockets) */
1226 } /* void network_send_buffer */
1227
1228 static int add_to_buffer (char *buffer, int buffer_size,
1229                 value_list_t *vl_def, char *type_def,
1230                 const data_set_t *ds, const value_list_t *vl)
1231 {
1232         char *buffer_orig = buffer;
1233
1234         if (strcmp (vl_def->host, vl->host) != 0)
1235         {
1236                 if (write_part_string (&buffer, &buffer_size, TYPE_HOST,
1237                                         vl->host, strlen (vl->host)) != 0)
1238                         return (-1);
1239                 strcpy (vl_def->host, vl->host);
1240         }
1241
1242         if (vl_def->time != vl->time)
1243         {
1244                 if (write_part_number (&buffer, &buffer_size, TYPE_TIME,
1245                                         (uint64_t) vl->time))
1246                         return (-1);
1247                 vl_def->time = vl->time;
1248         }
1249
1250         if (vl_def->interval != vl->interval)
1251         {
1252                 if (write_part_number (&buffer, &buffer_size, TYPE_INTERVAL,
1253                                         (uint64_t) vl->interval))
1254                         return (-1);
1255                 vl_def->interval = vl->interval;
1256         }
1257
1258         if (strcmp (vl_def->plugin, vl->plugin) != 0)
1259         {
1260                 if (write_part_string (&buffer, &buffer_size, TYPE_PLUGIN,
1261                                         vl->plugin, strlen (vl->plugin)) != 0)
1262                         return (-1);
1263                 strcpy (vl_def->plugin, vl->plugin);
1264         }
1265
1266         if (strcmp (vl_def->plugin_instance, vl->plugin_instance) != 0)
1267         {
1268                 if (write_part_string (&buffer, &buffer_size, TYPE_PLUGIN_INSTANCE,
1269                                         vl->plugin_instance,
1270                                         strlen (vl->plugin_instance)) != 0)
1271                         return (-1);
1272                 strcpy (vl_def->plugin_instance, vl->plugin_instance);
1273         }
1274
1275         if (strcmp (type_def, ds->type) != 0)
1276         {
1277                 if (write_part_string (&buffer, &buffer_size, TYPE_TYPE,
1278                                         ds->type, strlen (ds->type)) != 0)
1279                         return (-1);
1280                 strcpy (type_def, ds->type);
1281         }
1282
1283         if (strcmp (vl_def->type_instance, vl->type_instance) != 0)
1284         {
1285                 if (write_part_string (&buffer, &buffer_size, TYPE_TYPE_INSTANCE,
1286                                         vl->type_instance,
1287                                         strlen (vl->type_instance)) != 0)
1288                         return (-1);
1289                 strcpy (vl_def->type_instance, vl->type_instance);
1290         }
1291
1292         if (write_part_values (&buffer, &buffer_size, ds, vl) != 0)
1293                 return (-1);
1294
1295         return (buffer - buffer_orig);
1296 } /* int add_to_buffer */
1297
1298 static void flush_buffer (void)
1299 {
1300         DEBUG ("network plugin: flush_buffer: send_buffer_fill = %i",
1301                         send_buffer_fill);
1302
1303         network_send_buffer (send_buffer, send_buffer_fill);
1304         send_buffer_ptr  = send_buffer;
1305         send_buffer_fill = 0;
1306         memset (&send_buffer_vl, '\0', sizeof (send_buffer_vl));
1307         memset (send_buffer_type, '\0', sizeof (send_buffer_type));
1308 }
1309
1310 static int network_write (const data_set_t *ds, const value_list_t *vl)
1311 {
1312         int status;
1313
1314         /* If the value is already in the cache, we have received it via the
1315          * network. We write it again if forwarding is activated. It's then in
1316          * the cache and should we receive it again we will ignore it. */
1317         status = cache_check (ds->type, vl);
1318         if ((network_config_forward == 0)
1319                         && (status != 0))
1320                 return (0);
1321
1322         pthread_mutex_lock (&send_buffer_lock);
1323
1324         status = add_to_buffer (send_buffer_ptr,
1325                         sizeof (send_buffer) - send_buffer_fill,
1326                         &send_buffer_vl, send_buffer_type,
1327                         ds, vl);
1328         if (status >= 0)
1329         {
1330                 /* status == bytes added to the buffer */
1331                 send_buffer_fill += status;
1332                 send_buffer_ptr  += status;
1333         }
1334         else
1335         {
1336                 flush_buffer ();
1337
1338                 status = add_to_buffer (send_buffer_ptr,
1339                                 sizeof (send_buffer) - send_buffer_fill,
1340                                 &send_buffer_vl, send_buffer_type,
1341                                 ds, vl);
1342
1343                 if (status >= 0)
1344                 {
1345                         send_buffer_fill += status;
1346                         send_buffer_ptr  += status;
1347                 }
1348         }
1349
1350         if (status < 0)
1351         {
1352                 ERROR ("network plugin: Unable to append to the "
1353                                 "buffer for some weird reason");
1354         }
1355         else if ((sizeof (send_buffer) - send_buffer_fill) < 15)
1356         {
1357                 flush_buffer ();
1358         }
1359
1360         pthread_mutex_unlock (&send_buffer_lock);
1361
1362         return ((status < 0) ? -1 : 0);
1363 } /* int network_write */
1364
1365 static int network_config (const char *key, const char *val)
1366 {
1367         char *node;
1368         char *service;
1369
1370         char *fields[3];
1371         int   fields_num;
1372
1373         if ((strcasecmp ("Listen", key) == 0)
1374                         || (strcasecmp ("Server", key) == 0))
1375         {
1376                 char *val_cpy = strdup (val);
1377                 if (val_cpy == NULL)
1378                         return (1);
1379
1380                 service = NET_DEFAULT_PORT;
1381                 fields_num = strsplit (val_cpy, fields, 3);
1382                 if ((fields_num != 1)
1383                                 && (fields_num != 2))
1384                         return (1);
1385                 else if (fields_num == 2)
1386                 {
1387                         if ((service = strchr (fields[1], '.')) != NULL)
1388                                 *service = '\0';
1389                         service = fields[1];
1390                 }
1391                 node = fields[0];
1392
1393                 if (strcasecmp ("Listen", key) == 0)
1394                         network_add_listen_socket (node, service);
1395                 else
1396                         network_add_sending_socket (node, service);
1397         }
1398         else if (strcasecmp ("TimeToLive", key) == 0)
1399         {
1400                 int tmp = atoi (val);
1401                 if ((tmp > 0) && (tmp < 256))
1402                         network_config_ttl = tmp;
1403                 else
1404                         return (1);
1405         }
1406         else if (strcasecmp ("Forward", key) == 0)
1407         {
1408                 if ((strcasecmp ("true", val) == 0)
1409                                 || (strcasecmp ("yes", val) == 0)
1410                                 || (strcasecmp ("on", val) == 0))
1411                         network_config_forward = 1;
1412                 else
1413                         network_config_forward = 0;
1414         }
1415         else if (strcasecmp ("CacheFlush", key) == 0)
1416         {
1417                 int tmp = atoi (val);
1418                 if (tmp > 0)
1419                         cache_flush_interval = tmp;
1420                 else return (1);
1421         }
1422         else
1423         {
1424                 return (-1);
1425         }
1426         return (0);
1427 } /* int network_config */
1428
1429 static int network_shutdown (void)
1430 {
1431         listen_loop++;
1432
1433         /* Kill the listening thread */
1434         if (receive_thread_id != (pthread_t) 0)
1435         {
1436                 pthread_kill (receive_thread_id, SIGTERM);
1437                 pthread_join (receive_thread_id, NULL /* no return value */);
1438                 receive_thread_id = (pthread_t) 0;
1439         }
1440
1441         /* Shutdown the dispatching thread */
1442         if (dispatch_thread_id != (pthread_t) 0)
1443                 pthread_cond_broadcast (&receive_list_cond);
1444
1445         if (send_buffer_fill > 0)
1446                 flush_buffer ();
1447
1448         if (cache_tree != NULL)
1449         {
1450                 void *key;
1451                 void *value;
1452
1453                 while (c_avl_pick (cache_tree, &key, &value) == 0)
1454                 {
1455                         sfree (key);
1456                         sfree (value);
1457                 }
1458                 c_avl_destroy (cache_tree);
1459                 cache_tree = NULL;
1460         }
1461
1462         /* TODO: Close `sending_sockets' */
1463
1464         plugin_unregister_config ("network");
1465         plugin_unregister_init ("network");
1466         plugin_unregister_write ("network");
1467         plugin_unregister_shutdown ("network");
1468
1469         return (0);
1470 } /* int network_shutdown */
1471
1472 static int network_init (void)
1473 {
1474         plugin_register_shutdown ("network", network_shutdown);
1475
1476         send_buffer_ptr  = send_buffer;
1477         send_buffer_fill = 0;
1478         memset (&send_buffer_vl, '\0', sizeof (send_buffer_vl));
1479         memset (send_buffer_type, '\0', sizeof (send_buffer_type));
1480
1481         cache_tree = c_avl_create ((int (*) (const void *, const void *)) strcmp);
1482         cache_flush_last = time (NULL);
1483
1484         /* setup socket(s) and so on */
1485         if (sending_sockets != NULL)
1486                 plugin_register_write ("network", network_write);
1487
1488         if ((listen_sockets_num != 0) && (receive_thread_id == 0))
1489         {
1490                 int status;
1491
1492                 status = pthread_create (&dispatch_thread_id,
1493                                 NULL /* no attributes */,
1494                                 dispatch_thread,
1495                                 NULL /* no argument */);
1496                 if (status != 0)
1497                 {
1498                         char errbuf[1024];
1499                         ERROR ("network: pthread_create failed: %s",
1500                                         sstrerror (errno, errbuf,
1501                                                 sizeof (errbuf)));
1502                 }
1503
1504                 status = pthread_create (&receive_thread_id,
1505                                 NULL /* no attributes */,
1506                                 receive_thread,
1507                                 NULL /* no argument */);
1508                 if (status != 0)
1509                 {
1510                         char errbuf[1024];
1511                         ERROR ("network: pthread_create failed: %s",
1512                                         sstrerror (errno, errbuf,
1513                                                 sizeof (errbuf)));
1514                 }
1515         }
1516         return (0);
1517 } /* int network_init */
1518
1519 void module_register (void)
1520 {
1521         plugin_register_config ("network", network_config,
1522                         config_keys, config_keys_num);
1523         plugin_register_init   ("network", network_init);
1524 } /* void module_register */