src/erlang.c

   1 /**
   2  * collectd - src/erlang.c
   3  * Copyright (C) 2009  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
  25 #include <sys/types.h>
  26 #include <sys/socket.h>
  27 #include <netinet/in.h>
  28 #include <netdb.h>
  29
  30 #include <pthread.h>
  31
  32 #include <erl_interface.h>
  33 #include <ei.h>
  34
  35 /*
  36  * Private data structures
  37  */
  38 struct ce_connection_info_s
  39 {
  40         int fd;
  41         ErlConnect conn;
  42 };
  43 typedef struct ce_connection_info_s ce_connection_info_t;
  44
  45 /*
  46  * Private variables
  47  */
  48 static pthread_t listen_thread_id;
  49 static _Bool     listen_thread_running = false;
  50
  51 static char conf_service[NI_MAXSERV] = "29157";
  52 static char conf_cookie[256] = "ceisaequ";
  53
  54 /*
  55  * Private functions
  56  */
  57 static int send_atom (int fd, ETERM *to, const char *atom) /* {{{ */
  58 {
  59         ETERM *reply;
  60         int status;
  61
  62         reply = erl_mk_atom (atom);
  63         if (reply == NULL)
  64                 return (ENOMEM);
  65
  66         status = erl_send (fd, to, reply);
  67         erl_free_term (reply);
  68
  69         if (status == 1)
  70                 return (0);
  71         else
  72                 return (erl_errno);
  73 } /* }}} int send_atom */
  74
  75 static int send_error (int fd, ETERM *to, const char *message) /* {{{ */
  76 {
  77         ETERM *reply;
  78         int status;
  79
  80         DEBUG ("erlang plugin: send_error: message = %s.", message);
  81         reply = erl_format ("{~a,~s}", "error", message);
  82
  83         status = erl_send (fd, to, reply);
  84         if (status != 1)
  85                 status = erl_errno;
  86         else
  87                 status = 0;
  88
  89         erl_free_term (reply);
  90
  91         return (status);
  92 } /* }}} int send_error */
  93
  94 static int eterm_to_int (const ETERM *term, int *ret_int) /* {{{ */
  95 {
  96         if ((term == NULL) || (ret_int == NULL))
  97                 return (EINVAL);
  98
  99         switch (ERL_TYPE (term))
 100         {
 101                 case ERL_INTEGER:
 102                         *ret_int = (int) ERL_INT_VALUE (term);
 103                         break;
 104
 105                 case ERL_U_INTEGER:
 106                         *ret_int = (int) ERL_INT_UVALUE (term);
 107                         break;
 108
 109                 case ERL_FLOAT:
 110                         *ret_int = (int) (ERL_FLOAT_VALUE (term) + .5);
 111                         break;
 112
 113                 case ERL_LONGLONG:
 114                         *ret_int = (int) ERL_LL_VALUE (term);
 115                         break;
 116
 117                 case ERL_U_LONGLONG:
 118                         *ret_int = (int) ERL_LL_UVALUE (term);
 119                         break;
 120
 121                 default:
 122                         ERROR ("erlang plugin: Don't know how to cast "
 123                                         "erlang type %#x to int.", (unsigned int) ERL_TYPE (term));
 124                         return (ENOTSUP);
 125         } /* switch (ERL_TYPE (term)) */
 126
 127         return (0);
 128 } /* }}} int eterm_to_int */
 129
 130 static int eterm_to_time_t (const ETERM *term, time_t *ret_time) /* {{{ */
 131 {
 132         if ((term == NULL) || (ret_time == NULL))
 133                 return (EINVAL);
 134
 135         if (ERL_IS_NIL (term)
 136                         || (ERL_IS_ATOM (term)
 137                                 && ((strcmp ("now", ERL_ATOM_PTR (term)) == 0)
 138                                         || (strcmp ("undefined", ERL_ATOM_PTR (term)) == 0))))
 139         {
 140                 *ret_time = time (NULL);
 141                 return (0);
 142         }
 143
 144         switch (ERL_TYPE (term))
 145         {
 146                 case ERL_INTEGER:
 147                         *ret_time = (time_t) ERL_INT_VALUE (term);
 148                         break;
 149
 150                 case ERL_U_INTEGER:
 151                         *ret_time = (time_t) ERL_INT_UVALUE (term);
 152                         break;
 153
 154                 case ERL_ATOM:
 155                         if ((strcmp ("now", ERL_ATOM_PTR (term)) == 0)
 156                                         || (strcmp ("undefined", ERL_ATOM_PTR (term)) == 0))
 157                         {
 158                                 *ret_time = time (NULL);
 159                         }
 160                         else
 161                         {
 162                                 ERROR ("erlang plugin: Invalid atom for time: %s.",
 163                                                 ERL_ATOM_PTR (term));
 164                                 return (ENOTSUP);
 165                         }
 166                         break;
 167
 168                 case ERL_FLOAT:
 169                         *ret_time = (time_t) (ERL_FLOAT_VALUE (term) + .5);
 170                         break;
 171
 172                 case ERL_LONGLONG:
 173                         *ret_time = (time_t) ERL_LL_VALUE (term);
 174                         break;
 175
 176                 case ERL_U_LONGLONG:
 177                         *ret_time = (time_t) ERL_LL_UVALUE (term);
 178                         break;
 179
 180                 default:
 181                         ERROR ("erlang plugin: Don't know how to cast "
 182                                         "erlang type %#x to time_t.", (unsigned int) ERL_TYPE (term));
 183                         return (ENOTSUP);
 184         } /* switch (ERL_TYPE (term)) */
 185
 186         return (0);
 187 } /* }}} int eterm_to_time_t */
 188
 189 static int eterm_to_string (const ETERM *term, char *buffer, size_t buffer_size) /* {{{ */
 190 {
 191         char *tmp;
 192
 193         if ((term == NULL) || (buffer == NULL) || (buffer_size <= 0))
 194                 return (EINVAL);
 195
 196         memset (buffer, 0, buffer_size);
 197
 198         if (ERL_IS_EMPTY_LIST (term)
 199                         || ERL_IS_NIL (term)
 200                         || (ERL_IS_ATOM (term)
 201                                 && (strcmp ("undefined", ERL_ATOM_PTR (term)) == 0)))
 202         {
 203                 buffer[0] = 0;
 204                 return (0);
 205         }
 206
 207         if (!ERL_IS_LIST (term))
 208                 return (-1);
 209
 210         tmp = erl_iolist_to_string (term);
 211         if (tmp == NULL)
 212                 return (-1);
 213
 214         strncpy (buffer, tmp, buffer_size - 1);
 215         erl_free (tmp);
 216
 217         return (0);
 218 } /* }}} int eterm_to_string */
 219
 220 static int eterm_to_value (const ETERM *term, int ds_type, /* {{{ */
 221                 value_t *value)
 222 {
 223         if ((term == NULL) || (value == NULL))
 224                 return (EINVAL);
 225
 226         switch (ERL_TYPE (term))
 227         {
 228                 case ERL_INTEGER:
 229                 {
 230                         int v = ERL_INT_VALUE (term);
 231                         switch (ds_type)
 232                         {
 233                                 case DS_TYPE_COUNTER:  value->counter  = (counter_t)  v; break;
 234                                 case DS_TYPE_GAUGE:    value->gauge    = (gauge_t)    v; break;
 235                                 case DS_TYPE_DERIVE:   value->derive   = (derive_t)   v; break;
 236                                 case DS_TYPE_ABSOLUTE: value->absolute = (absolute_t) v; break;
 237                         }
 238                         break;
 239                 }
 240
 241                 case ERL_U_INTEGER:
 242                 {
 243                         unsigned int v = ERL_INT_UVALUE (term);
 244                         switch (ds_type)
 245                         {
 246                                 case DS_TYPE_COUNTER:  value->counter  = (counter_t)  v; break;
 247                                 case DS_TYPE_GAUGE:    value->gauge    = (gauge_t)    v; break;
 248                                 case DS_TYPE_DERIVE:   value->derive   = (derive_t)   v; break;
 249                                 case DS_TYPE_ABSOLUTE: value->absolute = (absolute_t) v; break;
 250                         }
 251                         break;
 252                 }
 253
 254                 case ERL_FLOAT:
 255                 {
 256                         double v = ERL_FLOAT_VALUE (term);
 257                         switch (ds_type)
 258                         {
 259                                 case DS_TYPE_COUNTER:  value->counter  = (counter_t)  v; break;
 260                                 case DS_TYPE_GAUGE:    value->gauge    = (gauge_t)    v; break;
 261                                 case DS_TYPE_DERIVE:   value->derive   = (derive_t)   v; break;
 262                                 case DS_TYPE_ABSOLUTE: value->absolute = (absolute_t) v; break;
 263                         }
 264                         break;
 265                 }
 266
 267                 case ERL_LONGLONG:
 268                 {
 269                         long long v = ERL_LL_VALUE (term);
 270                         switch (ds_type)
 271                         {
 272                                 case DS_TYPE_COUNTER:  value->counter  = (counter_t)  v; break;
 273                                 case DS_TYPE_GAUGE:    value->gauge    = (gauge_t)    v; break;
 274                                 case DS_TYPE_DERIVE:   value->derive   = (derive_t)   v; break;
 275                                 case DS_TYPE_ABSOLUTE: value->absolute = (absolute_t) v; break;
 276                         }
 277                         break;
 278                 }
 279
 280                 case ERL_U_LONGLONG:
 281                 {
 282                         unsigned long long v = ERL_LL_UVALUE (term);
 283                         switch (ds_type)
 284                         {
 285                                 case DS_TYPE_COUNTER:  value->counter  = (counter_t)  v; break;
 286                                 case DS_TYPE_GAUGE:    value->gauge    = (gauge_t)    v; break;
 287                                 case DS_TYPE_DERIVE:   value->derive   = (derive_t)   v; break;
 288                                 case DS_TYPE_ABSOLUTE: value->absolute = (absolute_t) v; break;
 289                         }
 290                         break;
 291                 }
 292
 293                 default:
 294                         ERROR ("erlang plugin: Don't know how to cast "
 295                                         "erlang type %#x to value_t.", (unsigned int) ERL_TYPE (term));
 296                         return (ENOTSUP);
 297         } /* switch (ERL_TYPE (term)) */
 298
 299         return (0);
 300 } /* }}} int eterm_to_value */
 301
 302 static int eterm_to_values (const ETERM *term, const data_set_t *ds, /* {{{ */
 303                 value_list_t *vl)
 304 {
 305         int ds_index;
 306         int status;
 307
 308         if ((term == NULL) || (ds == NULL) || (vl == NULL))
 309                 return (EINVAL);
 310
 311         if (!ERL_IS_LIST (term))
 312                 return (-1);
 313
 314         free (vl->values);
 315         vl->values = NULL;
 316         vl->values_len = 0;
 317
 318         while (!ERL_IS_EMPTY_LIST (term))
 319         {
 320                 const ETERM *eterm_value;
 321                 value_t *tmp;
 322
 323                 if (ds_index >= ds->ds_num)
 324                 {
 325                         ds_index = ds->ds_num + 1;
 326                         status = 0;
 327                         break;
 328                 }
 329
 330                 tmp = realloc (vl->values, sizeof (*tmp) * (vl->values_len + 1));
 331                 if (tmp == NULL)
 332                 {
 333                         status = ENOMEM;
 334                         break;
 335                 }
 336                 vl->values = tmp;
 337
 338                 eterm_value = ERL_CONS_HEAD (term);
 339                 term = ERL_CONS_TAIL (term);
 340
 341                 status = eterm_to_value (eterm_value, ds->ds[ds_index].type,
 342                                 vl->values + vl->values_len);
 343                 if (status != 0)
 344                         break;
 345
 346                 vl->values_len++;
 347                 ds_index++;
 348         }
 349
 350         if ((status == 0) && (ds_index != ds->ds_num))
 351                 NOTICE ("erlang plugin: Incorrect number of values received for type %s: "
 352                                 "Expected %i, got %i.", ds->type, ds->ds_num, ds_index);
 353
 354         if ((status != 0) || (ds_index != ds->ds_num))
 355         {
 356                 free (vl->values);
 357                 vl->values = NULL;
 358                 vl->values_len = 0;
 359                 return (status);
 360         }
 361
 362         return (0);
 363 } /* }}} int eterm_to_values */
 364
 365 static int eterm_to_value_list (const ETERM *term, value_list_t *vl) /* {{{ */
 366 {
 367         ETERM *tmp;
 368         int status;
 369         const data_set_t *ds;
 370
 371         if ((term == NULL) || (vl == NULL))
 372                 return (EINVAL);
 373
 374         if (!ERL_IS_TUPLE (term) || (ERL_TUPLE_SIZE (term) != 9))
 375                 return (EINVAL);
 376
 377         tmp = erl_element (1, term);
 378         if (!ERL_IS_ATOM (tmp)
 379                         || (strcmp ("value_list", ERL_ATOM_PTR (tmp)) != 0))
 380         {
 381                 erl_free_term (tmp);
 382                 return (-1);
 383         }
 384         erl_free_term (tmp);
 385
 386         status = 0;
 387         do
 388         {
 389 #define TUPLE_ELEM_TO_CHAR_ARRAY(idx,buf) \
 390                 tmp = erl_element ((idx), term); \
 391                 status = eterm_to_string (tmp, (buf), sizeof (buf)); \
 392                 erl_free_term (tmp); \
 393                 if (status != 0) \
 394                         break;
 395
 396                 TUPLE_ELEM_TO_CHAR_ARRAY (2, vl->host);
 397                 TUPLE_ELEM_TO_CHAR_ARRAY (3, vl->plugin);
 398                 TUPLE_ELEM_TO_CHAR_ARRAY (4, vl->plugin_instance);
 399                 TUPLE_ELEM_TO_CHAR_ARRAY (5, vl->type);
 400                 TUPLE_ELEM_TO_CHAR_ARRAY (6, vl->type_instance);
 401
 402                 ds = plugin_get_ds (vl->type);
 403                 if (ds == NULL)
 404                 {
 405                         status = -1;
 406                         break;
 407                 }
 408
 409                 tmp = erl_element (7, term);
 410                 status = eterm_to_time_t (tmp, &vl->time);
 411                 erl_free_term (tmp);
 412                 if (status != 0)
 413                         break;
 414
 415                 tmp = erl_element (8, term);
 416                 status = eterm_to_int (tmp, &vl->interval);
 417                 erl_free_term (tmp);
 418                 if (status != 0)
 419                         break;
 420                 if (vl->interval < 1)
 421                         vl->interval = interval_g;
 422
 423                 tmp = erl_element (9, term);
 424                 status = eterm_to_values (tmp, ds, vl);
 425                 erl_free_term (tmp);
 426                 if (status != 0)
 427                         break;
 428
 429 #undef TUPLE_ELEM_TO_CHAR_ARRAY
 430         } while (0);
 431
 432         if (status != 0)
 433                 return (status);
 434
 435         /* validate the struct */
 436         if ((vl->host[0] == 0) || (vl->plugin[0] == 0) || (vl->type[0] == 0))
 437                 return (-1);
 438
 439         if (ds->ds_num != vl->values_len)
 440                 return (-1);
 441
 442         return (0);
 443 } /* }}} int eterm_to_value_list */
 444
 445 /* Returns non-zero only if the request could not be handled gracefully. */
 446 static int handle_dispatch_values (ce_connection_info_t *cinfo, /* {{{ */
 447                 const ErlMessage *req)
 448 {
 449         ETERM *eterm_vl;
 450         value_list_t vl;
 451         int status;
 452
 453         memset (&vl, 0, sizeof (vl));
 454         vl.values = NULL;
 455
 456         eterm_vl = erl_element (2, req->msg);
 457         status = eterm_to_value_list (eterm_vl, &vl);
 458         erl_free_term (eterm_vl);
 459
 460         if (status != 0)
 461         {
 462                 free (vl.values);
 463                 send_error (cinfo->fd, req->from, "Cannot parse argument as value list.");
 464                 return (0);
 465         }
 466
 467         status = plugin_dispatch_values (&vl);
 468         if (status != 0)
 469         {
 470                 free (vl.values);
 471                 send_error (cinfo->fd, req->from, "plugin_dispatch_values failed.");
 472                 return (0);
 473         }
 474
 475         free (vl.values);
 476         send_atom (cinfo->fd, req->from, "success");
 477
 478         return (0);
 479 } /* }}} int handle_dispatch_values */
 480
 481 static void *handle_client_thread (void *arg) /* {{{ */
 482 {
 483         ce_connection_info_t *cinfo;
 484         ErlMessage emsg;
 485         unsigned char buffer[4096];
 486
 487         cinfo = arg;
 488
 489         DEBUG ("erlang plugin: handle_client_thread[%i]: Handling client %s.",
 490                         cinfo->fd, cinfo->conn.nodename);
 491
 492         emsg.from = NULL;
 493         emsg.to = NULL;
 494         emsg.msg = NULL;
 495
 496         while (42)
 497         {
 498                 int status;
 499
 500                 erl_free_term (emsg.from);
 501                 emsg.from = NULL;
 502                 erl_free_term (emsg.to);
 503                 emsg.to = NULL;
 504                 erl_free_term (emsg.msg);
 505                 emsg.msg = NULL;
 506
 507                 status = erl_receive_msg (cinfo->fd, buffer, sizeof (buffer), &emsg);
 508                 if (status == ERL_TICK)
 509                         continue;
 510
 511                 if (status == ERL_ERROR)
 512                         break;
 513
 514                 if (emsg.type == ERL_REG_SEND)
 515                 {
 516                         ETERM *func;
 517                         ETERM *reply;
 518
 519                         if (!ERL_IS_TUPLE (emsg.msg))
 520                         {
 521                                 ERROR ("erlang plugin: Message is not a tuple.");
 522                                 send_atom (cinfo->fd, emsg.from, "error");
 523                                 continue;
 524                         }
 525
 526                         func = erl_element (1, emsg.msg);
 527                         if (!ERL_IS_ATOM (func))
 528                         {
 529                                 ERROR ("erlang plugin: First element is not an atom!");
 530                                 send_atom (cinfo->fd, emsg.from, "error");
 531                                 erl_free_term (func);
 532                                 continue;
 533                         }
 534
 535                         DEBUG ("erlang plugin: Wanted function is: %s.", ERL_ATOM_PTR (func));
 536                         reply = NULL;
 537                         if (strcmp ("dispatch_values", ERL_ATOM_PTR (func)) == 0)
 538                                 status = handle_dispatch_values (cinfo, &emsg);
 539                         else
 540                         {
 541                                 ERROR ("erlang plugin: Received request for invalid function `%s'.",
 542                                                 ERL_ATOM_PTR (func));
 543                                 send_atom (cinfo->fd, emsg.from, "error");
 544                                 status = 0;
 545                         }
 546
 547                         /* Check for fatal errors in the callback functions. */
 548                         if (status != 0)
 549                         {
 550                                 ERROR ("erlang plugin: Handling request for `%s' failed.",
 551                                                 ERL_ATOM_PTR (func));
 552                                 erl_free_term (func);
 553                                 break;
 554                         }
 555
 556                         erl_free_term (func);
 557                 }
 558                 else if (emsg.type == ERL_EXIT)
 559                 {
 560                         DEBUG ("erlang plugin: handle_client_thread[%i]: "
 561                                         "Received exit message.", cinfo->fd);
 562                         break;
 563                 }
 564                 else
 565                 {
 566                         ERROR ("erlang plugin: Message type not handled: %i.", emsg.type);
 567                 }
 568         } /* while (42) */
 569
 570         erl_free_term (emsg.from);
 571         emsg.from = NULL;
 572         erl_free_term (emsg.to);
 573         emsg.to = NULL;
 574         erl_free_term (emsg.msg);
 575         emsg.msg = NULL;
 576
 577         DEBUG ("erlang plugin: handle_client_thread[%i]: Exiting.", cinfo->fd);
 578
 579         close (cinfo->fd);
 580         free (cinfo);
 581
 582         pthread_exit ((void *) 0);
 583         return ((void *) 0);
 584 } /* }}} void *handle_client_thread */
 585
 586 static int create_listen_socket (void) /* {{{ */
 587 {
 588         struct addrinfo ai_hints;
 589         struct addrinfo *ai_list;
 590         struct addrinfo *ai_ptr;
 591         int sock_descr;
 592         int status;
 593         int numeric_serv;
 594
 595         sock_descr = -1;
 596
 597         memset (&ai_hints, 0, sizeof (ai_hints));
 598         /* AI_PASSIVE => returns INADDR_ANY */
 599         ai_hints.ai_flags = AI_PASSIVE;
 600 #ifdef AI_ADDRCONFIG
 601         ai_hints.ai_flags |= AI_ADDRCONFIG;
 602 #endif
 603         /* IPv4 only :( */
 604         ai_hints.ai_family = AF_INET;
 605         ai_hints.ai_socktype = SOCK_STREAM;
 606
 607         ai_list = NULL;
 608         status = getaddrinfo (/* node = */ NULL, /* service = */ conf_service,
 609                         &ai_hints, &ai_list);
 610         if (status != 0)
 611         {
 612                 ERROR ("erlang plugin: getaddrinfo failed: %s", gai_strerror (status));
 613                 return (-1);
 614         }
 615
 616         for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
 617         {
 618                 struct sockaddr_in *sa_in;
 619                 struct in_addr *sin_addr;
 620                 int yes;
 621
 622                 assert (ai_ptr->ai_family == AF_INET);
 623                 sa_in = (struct sockaddr_in *) ai_ptr->ai_addr;
 624                 sin_addr = &sa_in->sin_addr;
 625                 numeric_serv = (int) ntohs (sa_in->sin_port);
 626
 627                 /* Dunno if calling this multiple times is legal. Since it wants to have
 628                  * the sin_addr for some reason this is the best place to call this,
 629                  * though. -octo */
 630                 status = erl_connect_xinit (/* host name = */ "leeloo",
 631                                 /* plain node name = */ "collectd",
 632                                 /* full node name  = */ "collectd@leeloo.lan.home.verplant.org",
 633                                 /* our address     = */ sin_addr,
 634                                 /* secret cookie   = */ conf_cookie,
 635                                 /* instance number = */ 0);
 636                 if (status < 0)
 637                 {
 638                         ERROR ("erlang plugin: erl_connect_xinit failed with status %i.",
 639                                         status);
 640                         continue;
 641                 }
 642
 643                 sock_descr = socket (ai_ptr->ai_family, ai_ptr->ai_socktype,
 644                                 ai_ptr->ai_protocol);
 645                 if (sock_descr < 0)
 646                 {
 647                         ERROR ("erlang plugin: socket(2) failed.");
 648                         continue;
 649                 }
 650
 651                 yes = 1;
 652                 status = setsockopt (sock_descr, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes));
 653                 if (status != 0)
 654                 {
 655                         ERROR ("erlang plugin: setsockopt(2) failed.");
 656                         close (sock_descr);
 657                         sock_descr = -1;
 658                         continue;
 659                 }
 660
 661                 status = bind (sock_descr, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
 662                 if (status != 0)
 663                 {
 664                         ERROR ("erlang plugin: bind(2) failed.");
 665                         close (sock_descr);
 666                         sock_descr = -1;
 667                         continue;
 668                 }
 669
 670                 status = listen (sock_descr, /* backlog = */ 10);
 671                 if (status != 0)
 672                 {
 673                         ERROR ("erlang plugin: listen(2) failed.");
 674                         close (sock_descr);
 675                         sock_descr = -1;
 676                         continue;
 677                 }
 678
 679                 break;
 680         } /* for (ai_list) */
 681
 682         freeaddrinfo (ai_list);
 683
 684         if (sock_descr >= 0)
 685         {
 686                 status = erl_publish (numeric_serv);
 687                 if (status < 0)
 688                 {
 689                         ERROR ("erlang plugin: erl_publish (%i) failed with status %i.", numeric_serv, status);
 690                         close (sock_descr);
 691                         sock_descr = -1;
 692                         return (-1);
 693                 }
 694         }
 695
 696         return (sock_descr);
 697 } /* }}} int create_listen_socket */
 698
 699 void *listen_thread (void *arg) /* {{{ */
 700 {
 701         int listen;
 702         int fd;
 703
 704         ErlConnect conn;
 705
 706         /* I have no fucking idea what this does, nor what the arguments are. Didn't
 707          * find any comprehensive docs yet. */
 708         erl_init (/* void *x = */ NULL, /* long y = */ 0);
 709
 710         listen = create_listen_socket ();
 711         if (listen < 0)
 712                 exit (EXIT_FAILURE);
 713
 714         while (42)
 715         {
 716                 pthread_t tid;
 717                 pthread_attr_t tattr;
 718                 ce_connection_info_t *arg;
 719
 720                 fd = erl_accept (listen, &conn);
 721                 if (fd < 0)
 722                 {
 723                         ERROR ("erlang plugin: erl_accept failed with status %i.", fd);
 724                         close (listen);
 725                         exit (EXIT_FAILURE);
 726                 }
 727                 DEBUG ("erlang plugin: Got connection from %s on fd %i.",
 728                                 conn.nodename, fd);
 729
 730                 pthread_attr_init (&tattr);
 731                 pthread_attr_setdetachstate (&tattr, PTHREAD_CREATE_DETACHED);
 732
 733                 arg = malloc (sizeof (*arg));
 734                 if (arg == NULL)
 735                 {
 736                         ERROR ("erlang plugin: malloc failed.");
 737                         close (fd);
 738                         continue;
 739                 }
 740                 memset (arg, 0, sizeof (*arg));
 741
 742                 arg->fd = fd;
 743                 memcpy (&arg->conn, &conn, sizeof (conn));
 744
 745                 pthread_create (&tid, &tattr, handle_client_thread, arg);
 746         } /* while (42) */
 747
 748         pthread_exit ((void *) 0);
 749         return ((void *) 0);
 750 } /* }}} void *listen_thread */
 751
 752 static int ce_init (void) /* {{{ */
 753 {
 754         if (!listen_thread_running)
 755         {
 756                 int status;
 757
 758                 status = pthread_create (&listen_thread_id,
 759                                 /* attr = */ NULL,
 760                                 listen_thread,
 761                                 /* args = */ NULL);
 762                 if (status == 0)
 763                         listen_thread_running = true;
 764         }
 765
 766         return (0);
 767 } /* }}} int ce_init */
 768
 769 void module_register (void)
 770 {
 771         plugin_register_init ("erlang", ce_init);
 772 }
 773
 774 /* vim: set sw=2 ts=2 noet fdm=marker : */