totemknet.c 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017
  1. /*
  2. * Copyright (c) 2016 Red Hat, Inc.
  3. *
  4. * All rights reserved.
  5. *
  6. * Author: Christine Caulfield (ccaulfie@redhat.com)
  7. * This software licensed under BSD license, the text of which follows:
  8. *
  9. * Redistribution and use in source and binary forms, with or without
  10. * modification, are permitted provided that the following conditions are met:
  11. *
  12. * - Redistributions of source code must retain the above copyright notice,
  13. * this list of conditions and the following disclaimer.
  14. * - Redistributions in binary form must reproduce the above copyright notice,
  15. * this list of conditions and the following disclaimer in the documentation
  16. * and/or other materials provided with the distribution.
  17. * - Neither the name of the MontaVista Software, Inc. nor the names of its
  18. * contributors may be used to endorse or promote products derived from this
  19. * software without specific prior written permission.
  20. *
  21. * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  22. * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  23. * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  24. * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
  25. * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  26. * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  27. * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  28. * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  29. * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  30. * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
  31. * THE POSSIBILITY OF SUCH DAMAGE.
  32. */
  33. #include <config.h>
  34. #include <assert.h>
  35. #include <sys/mman.h>
  36. #include <sys/types.h>
  37. #include <sys/stat.h>
  38. #include <sys/socket.h>
  39. #include <netdb.h>
  40. #include <sys/un.h>
  41. #include <sys/ioctl.h>
  42. #include <sys/param.h>
  43. #include <netinet/in.h>
  44. #include <arpa/inet.h>
  45. #include <unistd.h>
  46. #include <fcntl.h>
  47. #include <stdlib.h>
  48. #include <stdio.h>
  49. #include <errno.h>
  50. #include <sched.h>
  51. #include <time.h>
  52. #include <sys/time.h>
  53. #include <sys/poll.h>
  54. #include <sys/uio.h>
  55. #include <limits.h>
  56. #include <qb/qbdefs.h>
  57. #include <qb/qbloop.h>
  58. #include <corosync/sq.h>
  59. #include <corosync/swab.h>
  60. #include <corosync/logsys.h>
  61. #include <corosync/totem/totemip.h>
  62. #include "totemknet.h"
  63. #include "util.h"
  64. #include <nss.h>
  65. #include <pk11pub.h>
  66. #include <pkcs11.h>
  67. #include <prerror.h>
  68. #include <libknet.h>
  69. #ifndef MSG_NOSIGNAL
  70. #define MSG_NOSIGNAL 0
  71. #endif
  72. #define MCAST_SOCKET_BUFFER_SIZE (TRANSMITS_ALLOWED * FRAME_SIZE_MAX)
  73. /* Buffers for sendmmsg/recvmmsg */
  74. #define MAX_BUFFERS 10
  75. /* Should match that used by cfg */
  76. #define CFG_INTERFACE_STATUS_MAX_LEN 512
  77. /* Log messages received from libknet */
  78. #define LOG_BUFFER_SIZE 8192
  79. struct totemknet_instance {
  80. struct crypto_instance *crypto_inst;
  81. qb_loop_t *poll_handle;
  82. knet_handle_t knet_handle;
  83. int link_mode;
  84. void *context;
  85. void (*totemknet_deliver_fn) (
  86. void *context,
  87. const void *msg,
  88. unsigned int msg_len);
  89. void (*totemknet_iface_change_fn) (
  90. void *context,
  91. const struct totem_ip_address *iface_address,
  92. unsigned int link_no);
  93. void (*totemknet_mtu_changed) (
  94. void *context,
  95. int net_mtu);
  96. void (*totemknet_target_set_completed) (void *context);
  97. /*
  98. * Function and data used to log messages
  99. */
  100. int totemknet_log_level_security;
  101. int totemknet_log_level_error;
  102. int totemknet_log_level_warning;
  103. int totemknet_log_level_notice;
  104. int totemknet_log_level_debug;
  105. int totemknet_subsys_id;
  106. int knet_subsys_id;
  107. void (*totemknet_log_printf) (
  108. int level,
  109. int subsys,
  110. const char *function,
  111. const char *file,
  112. int line,
  113. const char *format,
  114. ...)__attribute__((format(printf, 6, 7)));
  115. void *knet_context;
  116. char iov_buffer[MAX_BUFFERS][FRAME_SIZE_MAX];
  117. int stats_sent;
  118. int stats_recv;
  119. int stats_delv;
  120. int stats_remcasts;
  121. int stats_orf_token;
  122. struct timeval stats_tv_start;
  123. char *link_status[INTERFACE_MAX];
  124. int num_links;
  125. struct totem_ip_address my_ids[INTERFACE_MAX];
  126. uint16_t ip_port[INTERFACE_MAX];
  127. int our_nodeid;
  128. struct totem_config *totem_config;
  129. totemsrp_stats_t *stats;
  130. struct totem_ip_address token_target;
  131. qb_loop_timer_handle timer_netif_check_timeout;
  132. qb_loop_timer_handle timer_merge_detect_timeout;
  133. int send_merge_detect_message;
  134. unsigned int merge_detect_messages_sent_before_timeout;
  135. int logpipes[2];
  136. int knet_fd;
  137. };
  138. struct work_item {
  139. const void *msg;
  140. unsigned int msg_len;
  141. struct totemknet_instance *instance;
  142. };
  143. int totemknet_member_list_rebind_ip (
  144. void *knet_context);
  145. static void totemknet_start_merge_detect_timeout(
  146. void *knet_context);
  147. static void totemknet_stop_merge_detect_timeout(
  148. void *knet_context);
  149. static void totemknet_instance_initialize (struct totemknet_instance *instance)
  150. {
  151. memset (instance, 0, sizeof (struct totemknet_instance));
  152. }
  153. #define knet_log_printf(level, format, args...) \
  154. do { \
  155. instance->totemknet_log_printf ( \
  156. level, instance->totemknet_subsys_id, \
  157. __FUNCTION__, __FILE__, __LINE__, \
  158. (const char *)format, ##args); \
  159. } while (0);
  160. #define libknet_log_printf(level, format, args...) \
  161. do { \
  162. instance->totemknet_log_printf ( \
  163. level, instance->knet_subsys_id, \
  164. __FUNCTION__, "libknet.h", __LINE__, \
  165. (const char *)format, ##args); \
  166. } while (0);
  167. #define KNET_LOGSYS_PERROR(err_num, level, fmt, args...) \
  168. do { \
  169. char _error_str[LOGSYS_MAX_PERROR_MSG_LEN]; \
  170. const char *_error_ptr = qb_strerror_r(err_num, _error_str, sizeof(_error_str)); \
  171. instance->totemknet_log_printf ( \
  172. level, instance->totemknet_subsys_id, \
  173. __FUNCTION__, __FILE__, __LINE__, \
  174. fmt ": %s (%d)", ##args, _error_ptr, err_num); \
  175. } while(0)
  176. static int dst_host_filter_callback_fn(void *private_data,
  177. const unsigned char *outdata,
  178. ssize_t outdata_len,
  179. uint8_t tx_rx,
  180. uint16_t this_host_id,
  181. uint16_t src_host_id,
  182. int8_t *channel,
  183. uint16_t *dst_host_ids,
  184. size_t *dst_host_ids_entries)
  185. {
  186. struct totem_message_header *header = (struct totem_message_header *)outdata;
  187. // struct totemknet_instance *instance = (struct totemknet_instance *)private_data;
  188. int res;
  189. // knet_log_printf (LOGSYS_LEVEL_DEBUG, "Filter notification called: %s target nodeid=%d, len=%ld", tx_rx==KNET_NOTIFY_RX?"RX":"TX", header->target_nodeid, outdata_len);
  190. *channel = 0;
  191. if (header->target_nodeid) {
  192. dst_host_ids[0] = header->target_nodeid;
  193. *dst_host_ids_entries = 1;
  194. res = 0; /* unicast message */
  195. }
  196. else {
  197. *dst_host_ids_entries = 0;
  198. res = 1; /* multicast message */
  199. }
  200. return res;
  201. }
  202. static void socket_error_callback_fn(void *private_data, int datafd, int8_t channel, uint8_t tx_rx, int error, int errorno)
  203. {
  204. struct totemknet_instance *instance = (struct totemknet_instance *)private_data;
  205. knet_log_printf (LOGSYS_LEVEL_DEBUG, "Knet socket ERROR notification called: txrx=%d, error=%d, errorno=%d", tx_rx, error, errorno);
  206. if ((error == -1 && errorno != EAGAIN) || (error == 0)) {
  207. knet_handle_remove_datafd(instance->knet_handle, datafd);
  208. }
  209. }
  210. static void host_change_callback_fn(void *private_data, uint16_t host_id, uint8_t reachable, uint8_t remote, uint8_t external)
  211. {
  212. struct totemknet_instance *instance = (struct totemknet_instance *)private_data;
  213. // TODO: what? if anything.
  214. knet_log_printf (LOGSYS_LEVEL_DEBUG, "Knet host change callback. nodeid: %d reachable: %d", host_id, reachable);
  215. }
  216. static void pmtu_change_callback_fn(void *private_data, unsigned int data_mtu)
  217. {
  218. struct totemknet_instance *instance = (struct totemknet_instance *)private_data;
  219. knet_log_printf (LOGSYS_LEVEL_DEBUG, "Knet pMTU change: %d", data_mtu);
  220. // TODO: Check this
  221. instance->totemknet_mtu_changed(instance->context, data_mtu - totemip_udpip_header_size(AF_INET));
  222. }
  223. int totemknet_crypto_set (
  224. void *knet_context,
  225. const char *cipher_type,
  226. const char *hash_type)
  227. {
  228. return (0);
  229. }
  230. static inline void ucast_sendmsg (
  231. struct totemknet_instance *instance,
  232. struct totem_ip_address *system_to,
  233. const void *msg,
  234. unsigned int msg_len)
  235. {
  236. int res = 0;
  237. struct totem_message_header *header = (struct totem_message_header *)msg;
  238. header->target_nodeid = system_to->nodeid;
  239. /*
  240. * Transmit unicast message
  241. * An error here is recovered by totemsrp
  242. */
  243. /*
  244. * If sending to ourself then just pass it through knet back to
  245. * the receive fn. knet does not do local->local delivery
  246. */
  247. if (system_to->nodeid == instance->our_nodeid) {
  248. res = write (instance->knet_fd+1, msg, msg_len);
  249. if (res < 0) {
  250. KNET_LOGSYS_PERROR (errno, instance->totemknet_log_level_debug,
  251. "sendmsg(ucast-local) failed (non-critical)");
  252. }
  253. }
  254. else {
  255. res = write (instance->knet_fd, msg, msg_len);
  256. if (res < 0) {
  257. KNET_LOGSYS_PERROR (errno, instance->totemknet_log_level_debug,
  258. "sendmsg(ucast) failed (non-critical)");
  259. }
  260. }
  261. }
  262. static inline void mcast_sendmsg (
  263. struct totemknet_instance *instance,
  264. const void *msg,
  265. unsigned int msg_len,
  266. int only_active)
  267. {
  268. int res;
  269. struct totem_message_header *header = (struct totem_message_header *)msg;
  270. header->target_nodeid = 0;
  271. // log_printf (LOGSYS_LEVEL_DEBUG, "totemknet: mcast_sendmsg. only_active=%d, len=%d", only_active, msg_len);
  272. res = write (instance->knet_fd, msg, msg_len);
  273. if (res < msg_len) {
  274. knet_log_printf (LOGSYS_LEVEL_DEBUG, "totemknet: mcast_send writev returned %d", res);
  275. }
  276. /*
  277. * Also send it to ourself, directly into
  278. * the receive fn. knet does not to local->local delivery
  279. */
  280. res = write (instance->knet_fd+1, msg, msg_len);
  281. if (res < msg_len) {
  282. knet_log_printf (LOGSYS_LEVEL_DEBUG, "totemknet: mcast_send writev (local) returned %d", res);
  283. }
  284. if (!only_active || instance->send_merge_detect_message) {
  285. /*
  286. * Current message was sent to all nodes
  287. */
  288. instance->merge_detect_messages_sent_before_timeout++;
  289. instance->send_merge_detect_message = 0;
  290. }
  291. }
  292. static int node_compare(const void *aptr, const void *bptr)
  293. {
  294. uint16_t a,b;
  295. a = *(uint16_t *)aptr;
  296. b = *(uint16_t *)bptr;
  297. return a > b;
  298. }
  299. int totemknet_ifaces_get (void *knet_context,
  300. char ***status,
  301. unsigned int *iface_count)
  302. {
  303. struct totemknet_instance *instance = (struct totemknet_instance *)knet_context;
  304. struct knet_link_status link_status;
  305. uint16_t host_list[KNET_MAX_HOST];
  306. size_t num_hosts;
  307. int i,j;
  308. char *ptr;
  309. int res = 0;
  310. /*
  311. * Don't do the whole 'link_info' bit if the caller just wants
  312. * a count of interfaces.
  313. */
  314. if (status) {
  315. res = knet_host_get_host_list(instance->knet_handle,
  316. host_list, &num_hosts);
  317. if (res) {
  318. return (-1);
  319. }
  320. qsort(host_list, num_hosts, sizeof(uint16_t), node_compare);
  321. /* num_links is actually the highest link ID */
  322. for (i=0; i <= instance->num_links; i++) {
  323. ptr = instance->link_status[i];
  324. for (j=0; j<num_hosts; j++) {
  325. res = knet_link_get_status(instance->knet_handle,
  326. host_list[j],
  327. i,
  328. &link_status);
  329. if (res == 0) {
  330. ptr[j] = '0' + (link_status.enabled |
  331. link_status.connected<<1 |
  332. link_status.dynconnected<<2);
  333. }
  334. else {
  335. ptr[j] += '?';
  336. }
  337. }
  338. ptr[num_hosts] = '\0';
  339. }
  340. *status = instance->link_status;
  341. }
  342. *iface_count = instance->num_links+1;
  343. return (res);
  344. }
  345. int totemknet_finalize (
  346. void *knet_context)
  347. {
  348. struct totemknet_instance *instance = (struct totemknet_instance *)knet_context;
  349. int res = 0;
  350. knet_log_printf(LOG_DEBUG, "totemknet: finalize");
  351. qb_loop_poll_del (instance->poll_handle, instance->logpipes[0]);
  352. qb_loop_poll_del (instance->poll_handle, instance->knet_fd);
  353. knet_handle_free(instance->knet_handle);
  354. totemknet_stop_merge_detect_timeout(instance);
  355. return (res);
  356. }
  357. static int log_deliver_fn (
  358. int fd,
  359. int revents,
  360. void *data)
  361. {
  362. struct totemknet_instance *instance = (struct totemknet_instance *)data;
  363. char buffer[LOG_BUFFER_SIZE];
  364. int len;
  365. len = read(fd, buffer, sizeof(buffer));
  366. if (len) {
  367. struct knet_log_msg *msg = (struct knet_log_msg *)buffer;
  368. switch (msg->msglevel) {
  369. case KNET_LOG_ERR:
  370. libknet_log_printf (LOGSYS_LEVEL_ERROR, "%s", msg->msg);
  371. break;
  372. case KNET_LOG_WARN:
  373. libknet_log_printf (LOGSYS_LEVEL_WARNING, "%s", msg->msg);
  374. break;
  375. case KNET_LOG_INFO:
  376. libknet_log_printf (LOGSYS_LEVEL_INFO, "%s", msg->msg);
  377. break;
  378. case KNET_LOG_DEBUG:
  379. libknet_log_printf (LOGSYS_LEVEL_DEBUG, "%s", msg->msg);
  380. break;
  381. }
  382. }
  383. return 0;
  384. }
  385. static int data_deliver_fn (
  386. int fd,
  387. int revents,
  388. void *data)
  389. {
  390. struct totemknet_instance *instance = (struct totemknet_instance *)data;
  391. struct mmsghdr msg_recv[MAX_BUFFERS];
  392. struct iovec iov_recv[MAX_BUFFERS];
  393. struct sockaddr_storage system_from;
  394. int msgs_received;
  395. int i;
  396. for (i=0; i<MAX_BUFFERS; i++) {
  397. iov_recv[i].iov_base = instance->iov_buffer[i];
  398. iov_recv[i].iov_len = FRAME_SIZE_MAX;
  399. msg_recv[i].msg_hdr.msg_name = &system_from;
  400. msg_recv[i].msg_hdr.msg_namelen = sizeof (struct sockaddr_storage);
  401. msg_recv[i].msg_hdr.msg_iov = &iov_recv[i];
  402. msg_recv[i].msg_hdr.msg_iovlen = 1;
  403. #ifdef HAVE_MSGHDR_CONTROL
  404. msg_recv[i].msg_hdr.msg_control = 0;
  405. #endif
  406. #ifdef HAVE_MSGHDR_CONTROLLEN
  407. msg_recv[i].msg_hdr.msg_controllen = 0;
  408. #endif
  409. #ifdef HAVE_MSGHDR_FLAGS
  410. msg_recv[i].msg_hdr.msg_flags = 0;
  411. #endif
  412. #ifdef HAVE_MSGHDR_ACCRIGHTS
  413. msg_recv[i].msg_hdr.msg_accrights = NULL;
  414. #endif
  415. #ifdef HAVE_MSGHDR_ACCRIGHTSLEN
  416. msg_recv[i].msg_hdr.msg_accrightslen = 0;
  417. #endif
  418. }
  419. msgs_received = recvmmsg (fd, msg_recv, MAX_BUFFERS, MSG_NOSIGNAL | MSG_DONTWAIT, NULL);
  420. if (msgs_received == -1) {
  421. return (0);
  422. }
  423. for (i=0; i<msgs_received; i++) {
  424. instance->stats_recv += msg_recv[i].msg_len;
  425. /*
  426. * Handle incoming message
  427. */
  428. instance->totemknet_deliver_fn (
  429. instance->context,
  430. instance->iov_buffer[i],
  431. msg_recv[i].msg_len);
  432. }
  433. return (0);
  434. }
  435. static void timer_function_netif_check_timeout (
  436. void *data)
  437. {
  438. struct totemknet_instance *instance = (struct totemknet_instance *)data;
  439. int i;
  440. for (i=0; i<instance->totem_config->interface_count; i++)
  441. instance->totemknet_iface_change_fn (instance->context,
  442. &instance->my_ids[i],
  443. i);
  444. }
  445. /*
  446. * Create an instance
  447. */
  448. int totemknet_initialize (
  449. qb_loop_t *poll_handle,
  450. void **knet_context,
  451. struct totem_config *totem_config,
  452. totemsrp_stats_t *stats,
  453. void *context,
  454. void (*deliver_fn) (
  455. void *context,
  456. const void *msg,
  457. unsigned int msg_len),
  458. void (*iface_change_fn) (
  459. void *context,
  460. const struct totem_ip_address *iface_address,
  461. unsigned int link_no),
  462. void (*mtu_changed) (
  463. void *context,
  464. int net_mtu),
  465. void (*target_set_completed) (
  466. void *context))
  467. {
  468. struct totemknet_instance *instance;
  469. int8_t channel=0;
  470. int res;
  471. int i;
  472. instance = malloc (sizeof (struct totemknet_instance));
  473. if (instance == NULL) {
  474. return (-1);
  475. }
  476. totemknet_instance_initialize (instance);
  477. instance->totem_config = totem_config;
  478. instance->stats = stats;
  479. /*
  480. * Configure logging
  481. */
  482. instance->totemknet_log_level_security = 1; //totem_config->totem_logging_configuration.log_level_security;
  483. instance->totemknet_log_level_error = totem_config->totem_logging_configuration.log_level_error;
  484. instance->totemknet_log_level_warning = totem_config->totem_logging_configuration.log_level_warning;
  485. instance->totemknet_log_level_notice = totem_config->totem_logging_configuration.log_level_notice;
  486. instance->totemknet_log_level_debug = totem_config->totem_logging_configuration.log_level_debug;
  487. instance->totemknet_subsys_id = totem_config->totem_logging_configuration.log_subsys_id;
  488. instance->totemknet_log_printf = totem_config->totem_logging_configuration.log_printf;
  489. instance->knet_subsys_id = _logsys_subsys_create("KNET", "libknet.h");
  490. /*
  491. * Initialize local variables for totemknet
  492. */
  493. instance->our_nodeid = instance->totem_config->node_id;
  494. for (i=0; i< instance->totem_config->interface_count; i++) {
  495. totemip_copy(&instance->my_ids[i], &totem_config->interfaces[i].bindnet);
  496. instance->my_ids[i].nodeid = instance->our_nodeid;
  497. instance->ip_port[i] = totem_config->interfaces[i].ip_port;
  498. /* Needed for totemsrp */
  499. totem_config->interfaces[i].boundto.nodeid = instance->our_nodeid;
  500. }
  501. instance->poll_handle = poll_handle;
  502. instance->context = context;
  503. instance->totemknet_deliver_fn = deliver_fn;
  504. instance->totemknet_iface_change_fn = iface_change_fn;
  505. instance->totemknet_mtu_changed = mtu_changed;
  506. instance->totemknet_target_set_completed = target_set_completed;
  507. pipe(instance->logpipes);
  508. fcntl(instance->logpipes[0], F_SETFL, O_NONBLOCK);
  509. fcntl(instance->logpipes[1], F_SETFL, O_NONBLOCK);
  510. instance->knet_handle = knet_handle_new(instance->totem_config->node_id, instance->logpipes[1], KNET_LOG_DEBUG);
  511. if (!instance->knet_handle) {
  512. KNET_LOGSYS_PERROR(errno, LOGSYS_LEVEL_CRIT, "knet_handle_new failed");
  513. return (-1);
  514. }
  515. res = knet_handle_pmtud_setfreq(instance->knet_handle, 5);
  516. if (res) {
  517. KNET_LOGSYS_PERROR(errno, LOGSYS_LEVEL_WARNING, "knet_handle_pmtud_setfreq failed");
  518. }
  519. res = knet_handle_enable_filter(instance->knet_handle, instance, dst_host_filter_callback_fn);
  520. if (res) {
  521. KNET_LOGSYS_PERROR(errno, LOGSYS_LEVEL_WARNING, "knet_handle_enable_filter failed");
  522. }
  523. res = knet_handle_enable_sock_notify(instance->knet_handle, instance, socket_error_callback_fn);
  524. if (res) {
  525. KNET_LOGSYS_PERROR(errno, LOGSYS_LEVEL_WARNING, "knet_handle_enable_sock_notify failed");
  526. }
  527. res = knet_host_enable_status_change_notify(instance->knet_handle, instance, host_change_callback_fn);
  528. if (res) {
  529. KNET_LOGSYS_PERROR(errno, LOGSYS_LEVEL_WARNING, "knet_host_enable_status_change_notify failed");
  530. }
  531. res = knet_handle_enable_pmtud_notify(instance->knet_handle, instance, pmtu_change_callback_fn);
  532. if (res) {
  533. KNET_LOGSYS_PERROR(errno, LOGSYS_LEVEL_WARNING, "knet_handle_enable_pmtud_notify failed");
  534. }
  535. /* Get an fd into knet */
  536. instance->knet_fd = 0;
  537. res = knet_handle_add_datafd(instance->knet_handle, &instance->knet_fd, &channel);
  538. if (res) {
  539. knet_log_printf(LOG_DEBUG, "knet_handle_add_datafd failed: %s", strerror(errno));
  540. return -1;
  541. }
  542. /* Enable crypto if requested */
  543. if (strcmp(instance->totem_config->crypto_cipher_type, "none") != 0) {
  544. struct knet_handle_crypto_cfg crypto_cfg;
  545. strcpy(crypto_cfg.crypto_model, "nss");
  546. strcpy(crypto_cfg.crypto_cipher_type, instance->totem_config->crypto_cipher_type);
  547. strcpy(crypto_cfg.crypto_hash_type, instance->totem_config->crypto_hash_type);
  548. memcpy(crypto_cfg.private_key, instance->totem_config->private_key, instance->totem_config->private_key_len);
  549. crypto_cfg.private_key_len = instance->totem_config->private_key_len;
  550. res = knet_handle_crypto(instance->knet_handle, &crypto_cfg);
  551. if (res == -1) {
  552. knet_log_printf(LOG_ERR, "knet_handle_crypto failed: %s", strerror(errno));
  553. return -1;
  554. }
  555. if (res == -2) {
  556. knet_log_printf(LOG_ERR, "knet_handle_crypto failed: -2");
  557. return -1;
  558. }
  559. knet_log_printf(LOG_INFO, "kronosnet crypto initialized: %s/%s", crypto_cfg.crypto_cipher_type, crypto_cfg.crypto_hash_type);
  560. }
  561. knet_handle_setfwd(instance->knet_handle, 1);
  562. instance->link_mode = KNET_LINK_POLICY_PASSIVE;
  563. if (strcmp(instance->totem_config->link_mode, "active")==0) {
  564. instance->link_mode = KNET_LINK_POLICY_ACTIVE;
  565. }
  566. if (strcmp(instance->totem_config->link_mode, "rr")==0) {
  567. instance->link_mode = KNET_LINK_POLICY_RR;
  568. }
  569. for (i=0; i<INTERFACE_MAX; i++) {
  570. instance->link_status[i] = malloc(CFG_INTERFACE_STATUS_MAX_LEN);
  571. if (!instance->link_status[i]) {
  572. return -1;
  573. }
  574. }
  575. qb_loop_poll_add (instance->poll_handle,
  576. QB_LOOP_MED,
  577. instance->logpipes[0],
  578. POLLIN, instance, log_deliver_fn);
  579. qb_loop_poll_add (instance->poll_handle,
  580. QB_LOOP_HIGH,
  581. instance->knet_fd,
  582. POLLIN, instance, data_deliver_fn);
  583. /*
  584. * Upper layer isn't ready to receive message because it hasn't
  585. * initialized yet. Add short timer to check the interfaces.
  586. */
  587. qb_loop_timer_add (instance->poll_handle,
  588. QB_LOOP_MED,
  589. 100*QB_TIME_NS_IN_MSEC,
  590. (void *)instance,
  591. timer_function_netif_check_timeout,
  592. &instance->timer_netif_check_timeout);
  593. totemknet_start_merge_detect_timeout(instance);
  594. knet_log_printf (LOGSYS_LEVEL_INFO, "totemknet initialized");
  595. *knet_context = instance;
  596. return (0);
  597. }
  598. void *totemknet_buffer_alloc (void)
  599. {
  600. return malloc (FRAME_SIZE_MAX);
  601. }
  602. void totemknet_buffer_release (void *ptr)
  603. {
  604. return free (ptr);
  605. }
  606. int totemknet_processor_count_set (
  607. void *knet_context,
  608. int processor_count)
  609. {
  610. return (0);
  611. }
  612. int totemknet_recv_flush (void *knet_context)
  613. {
  614. return (0);
  615. }
  616. int totemknet_send_flush (void *knet_context)
  617. {
  618. return (0);
  619. }
  620. int totemknet_token_send (
  621. void *knet_context,
  622. const void *msg,
  623. unsigned int msg_len)
  624. {
  625. struct totemknet_instance *instance = (struct totemknet_instance *)knet_context;
  626. int res = 0;
  627. ucast_sendmsg (instance, &instance->token_target, msg, msg_len);
  628. return (res);
  629. }
  630. int totemknet_mcast_flush_send (
  631. void *knet_context,
  632. const void *msg,
  633. unsigned int msg_len)
  634. {
  635. struct totemknet_instance *instance = (struct totemknet_instance *)knet_context;
  636. int res = 0;
  637. mcast_sendmsg (instance, msg, msg_len, 0);
  638. return (res);
  639. }
  640. int totemknet_mcast_noflush_send (
  641. void *knet_context,
  642. const void *msg,
  643. unsigned int msg_len)
  644. {
  645. struct totemknet_instance *instance = (struct totemknet_instance *)knet_context;
  646. int res = 0;
  647. mcast_sendmsg (instance, msg, msg_len, 1);
  648. return (res);
  649. }
  650. extern int totemknet_iface_check (void *knet_context)
  651. {
  652. struct totemknet_instance *instance = (struct totemknet_instance *)knet_context;
  653. int res = 0;
  654. knet_log_printf(LOG_DEBUG, "totmeknet: iface_check");
  655. return (res);
  656. }
  657. extern void totemknet_net_mtu_adjust (void *knet_context, struct totem_config *totem_config)
  658. {
  659. struct totemknet_instance *instance = (struct totemknet_instance *)knet_context;
  660. totem_config->net_mtu -= totemip_udpip_header_size(AF_INET) + 23;
  661. knet_log_printf(LOG_DEBUG, "totemknet: Returning MTU of %d", totem_config->net_mtu);
  662. }
  663. int totemknet_token_target_set (
  664. void *knet_context,
  665. const struct totem_ip_address *token_target)
  666. {
  667. struct totemknet_instance *instance = (struct totemknet_instance *)knet_context;
  668. int res = 0;
  669. memcpy (&instance->token_target, token_target,
  670. sizeof (struct totem_ip_address));
  671. instance->totemknet_target_set_completed (instance->context);
  672. return (res);
  673. }
  674. extern int totemknet_recv_mcast_empty (
  675. void *knet_context)
  676. {
  677. struct totemknet_instance *instance = (struct totemknet_instance *)knet_context;
  678. unsigned int res;
  679. struct sockaddr_storage system_from;
  680. struct mmsghdr msg_recv[MAX_BUFFERS];
  681. struct iovec iov_recv[MAX_BUFFERS];
  682. struct pollfd ufd;
  683. int nfds;
  684. int msg_processed = 0;
  685. int i;
  686. for (i=0; i<MAX_BUFFERS; i++) {
  687. iov_recv[i].iov_base = instance->iov_buffer[i];
  688. iov_recv[i].iov_len = FRAME_SIZE_MAX;
  689. msg_recv[i].msg_hdr.msg_name = &system_from;
  690. msg_recv[i].msg_hdr.msg_namelen = sizeof (struct sockaddr_storage);
  691. msg_recv[i].msg_hdr.msg_iov = &iov_recv[i];
  692. msg_recv[i].msg_hdr.msg_iovlen = 1;
  693. #ifdef HAVE_MSGHDR_CONTROL
  694. msg_recv[i].msg_hdr.msg_control = 0;
  695. #endif
  696. #ifdef HAVE_MSGHDR_CONTROLLEN
  697. msg_recv[i].msg_hdr.msg_controllen = 0;
  698. #endif
  699. #ifdef HAVE_MSGHDR_FLAGS
  700. msg_recv[i].msg_hdr.msg_flags = 0;
  701. #endif
  702. #ifdef HAVE_MSGHDR_ACCRIGHTS
  703. msg_recv[i].msg_hdr.msg_accrights = NULL;
  704. #endif
  705. #ifdef HAVE_MSGHDR_ACCRIGHTSLEN
  706. msg_recv[i].msg_hdr.msg_accrightslen = 0;
  707. #endif
  708. }
  709. do {
  710. ufd.fd = instance->knet_fd;
  711. ufd.events = POLLIN;
  712. nfds = poll (&ufd, 1, 0);
  713. if (nfds == 1 && ufd.revents & POLLIN) {
  714. res = recvmmsg (instance->knet_fd, msg_recv, MAX_BUFFERS, MSG_NOSIGNAL | MSG_DONTWAIT, NULL);
  715. if (res != -1) {
  716. msg_processed = 1;
  717. } else {
  718. msg_processed = -1;
  719. }
  720. }
  721. } while (nfds == 1);
  722. return (msg_processed);
  723. }
  724. int totemknet_member_add (
  725. void *knet_context,
  726. const struct totem_ip_address *local,
  727. const struct totem_ip_address *member,
  728. int link_no)
  729. {
  730. struct totemknet_instance *instance = (struct totemknet_instance *)knet_context;
  731. int err;
  732. int port = instance->ip_port[link_no];
  733. struct sockaddr_storage remote_ss;
  734. struct sockaddr_storage local_ss;
  735. int addrlen;
  736. if (member->nodeid == instance->our_nodeid) {
  737. return 0; /* Don't add ourself, we send loopback messages directly */
  738. }
  739. /* Keep track of the number of links */
  740. if (link_no > instance->num_links) {
  741. instance->num_links = link_no;
  742. }
  743. knet_log_printf (LOGSYS_LEVEL_DEBUG, "knet: member_add: %d (%s), link=%d", member->nodeid, totemip_print(member), link_no);
  744. knet_log_printf (LOGSYS_LEVEL_DEBUG, "knet: local: %d (%s)", local->nodeid, totemip_print(local));
  745. if (link_no == 0) {
  746. if (knet_host_add(instance->knet_handle, member->nodeid)) {
  747. KNET_LOGSYS_PERROR(errno, LOGSYS_LEVEL_ERROR, "knet_host_add");
  748. return -1;
  749. }
  750. if (knet_host_set_policy(instance->knet_handle, member->nodeid, instance->link_mode)) {
  751. KNET_LOGSYS_PERROR(errno, LOGSYS_LEVEL_ERROR, "knet_set_policy failed");
  752. return -1;
  753. }
  754. }
  755. /* Casts to remove const */
  756. totemip_totemip_to_sockaddr_convert((struct totem_ip_address *)member, port+link_no, &remote_ss, &addrlen);
  757. totemip_totemip_to_sockaddr_convert((struct totem_ip_address *)local, port+link_no, &local_ss, &addrlen);
  758. err = knet_link_set_config(instance->knet_handle, member->nodeid, link_no, &local_ss, &remote_ss);
  759. if (err) {
  760. KNET_LOGSYS_PERROR(errno, LOGSYS_LEVEL_ERROR, "knet_link_set_config failed");
  761. return -1;
  762. }
  763. knet_log_printf (LOGSYS_LEVEL_DEBUG, "knet: member_add: Setting link prio to %d",
  764. instance->totem_config->interfaces[link_no].knet_link_priority);
  765. err = knet_link_set_priority(instance->knet_handle, member->nodeid, link_no,
  766. instance->totem_config->interfaces[link_no].knet_link_priority);
  767. if (err) {
  768. KNET_LOGSYS_PERROR(errno, LOGSYS_LEVEL_ERROR, "knet_link_set_priority for nodeid %d, link %d failed", member->nodeid, link_no);
  769. }
  770. err = knet_link_set_ping_timers(instance->knet_handle, member->nodeid, link_no,
  771. instance->totem_config->interfaces[link_no].knet_ping_interval,
  772. instance->totem_config->interfaces[link_no].knet_ping_timeout,
  773. instance->totem_config->interfaces[link_no].knet_ping_precision);
  774. if (err) {
  775. KNET_LOGSYS_PERROR(errno, LOGSYS_LEVEL_ERROR, "knet_link_set_ping_timers for nodeid %d, link %d failed", member->nodeid, link_no);
  776. }
  777. err = knet_link_set_enable(instance->knet_handle, member->nodeid, link_no, 1);
  778. if (err) {
  779. KNET_LOGSYS_PERROR(errno, LOGSYS_LEVEL_ERROR, "knet_link_set_enable for nodeid %d, link %d failed", member->nodeid, link_no);
  780. return -1;
  781. }
  782. return (0);
  783. }
  784. int totemknet_member_remove (
  785. void *knet_context,
  786. const struct totem_ip_address *token_target,
  787. int link_no)
  788. {
  789. struct totemknet_instance *instance = (struct totemknet_instance *)knet_context;
  790. return knet_host_remove(instance->knet_handle, token_target->nodeid);
  791. }
  792. int totemknet_member_list_rebind_ip (
  793. void *knet_context)
  794. {
  795. return (0);
  796. }
  797. static void timer_function_merge_detect_timeout (
  798. void *data)
  799. {
  800. struct totemknet_instance *instance = (struct totemknet_instance *)data;
  801. if (instance->merge_detect_messages_sent_before_timeout == 0) {
  802. instance->send_merge_detect_message = 1;
  803. }
  804. instance->merge_detect_messages_sent_before_timeout = 0;
  805. totemknet_start_merge_detect_timeout(instance);
  806. }
  807. static void totemknet_start_merge_detect_timeout(
  808. void *knet_context)
  809. {
  810. struct totemknet_instance *instance = (struct totemknet_instance *)knet_context;
  811. qb_loop_timer_add(instance->poll_handle,
  812. QB_LOOP_MED,
  813. instance->totem_config->merge_timeout * 2 * QB_TIME_NS_IN_MSEC,
  814. (void *)instance,
  815. timer_function_merge_detect_timeout,
  816. &instance->timer_merge_detect_timeout);
  817. }
  818. static void totemknet_stop_merge_detect_timeout(
  819. void *knet_context)
  820. {
  821. struct totemknet_instance *instance = (struct totemknet_instance *)knet_context;
  822. qb_loop_timer_del(instance->poll_handle,
  823. instance->timer_merge_detect_timeout);
  824. }