lowcomms.c 42 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724
  1. // SPDX-License-Identifier: GPL-2.0-only
  2. /******************************************************************************
  3. *******************************************************************************
  4. **
  5. ** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
  6. ** Copyright (C) 2004-2009 Red Hat, Inc. All rights reserved.
  7. **
  8. **
  9. *******************************************************************************
  10. ******************************************************************************/
  11. /*
  12. * lowcomms.c
  13. *
  14. * This is the "low-level" comms layer.
  15. *
  16. * It is responsible for sending/receiving messages
  17. * from other nodes in the cluster.
  18. *
  19. * Cluster nodes are referred to by their nodeids. nodeids are
  20. * simply 32 bit numbers to the locking module - if they need to
  21. * be expanded for the cluster infrastructure then that is its
  22. * responsibility. It is this layer's
  23. * responsibility to resolve these into IP address or
  24. * whatever it needs for inter-node communication.
  25. *
  26. * The comms level is two kernel threads that deal mainly with
  27. * the receiving of messages from other nodes and passing them
  28. * up to the mid-level comms layer (which understands the
  29. * message format) for execution by the locking core, and
  30. * a send thread which does all the setting up of connections
  31. * to remote nodes and the sending of data. Threads are not allowed
  32. * to send their own data because it may cause them to wait in times
  33. * of high load. Also, this way, the sending thread can collect together
  34. * messages bound for one node and send them in one block.
  35. *
  36. * lowcomms will choose to use either TCP or SCTP as its transport layer
  37. * depending on the configuration variable 'protocol'. This should be set
  38. * to 0 (default) for TCP or 1 for SCTP. It should be configured using a
  39. * cluster-wide mechanism as it must be the same on all nodes of the cluster
  40. * for the DLM to function.
  41. *
  42. */
  43. #include <asm/ioctls.h>
  44. #include <net/sock.h>
  45. #include <net/tcp.h>
  46. #include <linux/pagemap.h>
  47. #include <linux/file.h>
  48. #include <linux/mutex.h>
  49. #include <linux/sctp.h>
  50. #include <linux/slab.h>
  51. #include <net/sctp/sctp.h>
  52. #include <net/ipv6.h>
  53. #include "dlm_internal.h"
  54. #include "lowcomms.h"
  55. #include "midcomms.h"
  56. #include "config.h"
  57. #define NEEDED_RMEM (4*1024*1024)
  58. #define CONN_HASH_SIZE 32
  59. /* Number of messages to send before rescheduling */
  60. #define MAX_SEND_MSG_COUNT 25
  61. #define DLM_SHUTDOWN_WAIT_TIMEOUT msecs_to_jiffies(10000)
  62. struct connection {
  63. struct socket *sock; /* NULL if not connected */
  64. uint32_t nodeid; /* So we know who we are in the list */
  65. struct mutex sock_mutex;
  66. unsigned long flags;
  67. #define CF_READ_PENDING 1
  68. #define CF_WRITE_PENDING 2
  69. #define CF_INIT_PENDING 4
  70. #define CF_IS_OTHERCON 5
  71. #define CF_CLOSE 6
  72. #define CF_APP_LIMITED 7
  73. #define CF_CLOSING 8
  74. #define CF_SHUTDOWN 9
  75. struct list_head writequeue; /* List of outgoing writequeue_entries */
  76. spinlock_t writequeue_lock;
  77. int (*rx_action) (struct connection *); /* What to do when active */
  78. void (*connect_action) (struct connection *); /* What to do to connect */
  79. void (*shutdown_action)(struct connection *con); /* What to do to shutdown */
  80. int retries;
  81. #define MAX_CONNECT_RETRIES 3
  82. struct hlist_node list;
  83. struct connection *othercon;
  84. struct work_struct rwork; /* Receive workqueue */
  85. struct work_struct swork; /* Send workqueue */
  86. wait_queue_head_t shutdown_wait; /* wait for graceful shutdown */
  87. unsigned char *rx_buf;
  88. int rx_buflen;
  89. int rx_leftover;
  90. struct rcu_head rcu;
  91. };
  92. #define sock2con(x) ((struct connection *)(x)->sk_user_data)
  93. /* An entry waiting to be sent */
  94. struct writequeue_entry {
  95. struct list_head list;
  96. struct page *page;
  97. int offset;
  98. int len;
  99. int end;
  100. int users;
  101. struct connection *con;
  102. };
  103. struct dlm_node_addr {
  104. struct list_head list;
  105. int nodeid;
  106. int addr_count;
  107. int curr_addr_index;
  108. struct sockaddr_storage *addr[DLM_MAX_ADDR_COUNT];
  109. };
  110. static struct listen_sock_callbacks {
  111. void (*sk_error_report)(struct sock *);
  112. void (*sk_data_ready)(struct sock *);
  113. void (*sk_state_change)(struct sock *);
  114. void (*sk_write_space)(struct sock *);
  115. } listen_sock;
  116. static LIST_HEAD(dlm_node_addrs);
  117. static DEFINE_SPINLOCK(dlm_node_addrs_spin);
  118. static struct sockaddr_storage *dlm_local_addr[DLM_MAX_ADDR_COUNT];
  119. static int dlm_local_count;
  120. static int dlm_allow_conn;
  121. /* Work queues */
  122. static struct workqueue_struct *recv_workqueue;
  123. static struct workqueue_struct *send_workqueue;
  124. static struct hlist_head connection_hash[CONN_HASH_SIZE];
  125. static DEFINE_SPINLOCK(connections_lock);
  126. DEFINE_STATIC_SRCU(connections_srcu);
  127. static void process_recv_sockets(struct work_struct *work);
  128. static void process_send_sockets(struct work_struct *work);
  129. /* This is deliberately very simple because most clusters have simple
  130. sequential nodeids, so we should be able to go straight to a connection
  131. struct in the array */
  132. static inline int nodeid_hash(int nodeid)
  133. {
  134. return nodeid & (CONN_HASH_SIZE-1);
  135. }
  136. static struct connection *__find_con(int nodeid)
  137. {
  138. int r, idx;
  139. struct connection *con;
  140. r = nodeid_hash(nodeid);
  141. idx = srcu_read_lock(&connections_srcu);
  142. hlist_for_each_entry_rcu(con, &connection_hash[r], list) {
  143. if (con->nodeid == nodeid) {
  144. srcu_read_unlock(&connections_srcu, idx);
  145. return con;
  146. }
  147. }
  148. srcu_read_unlock(&connections_srcu, idx);
  149. return NULL;
  150. }
  151. /*
  152. * If 'allocation' is zero then we don't attempt to create a new
  153. * connection structure for this node.
  154. */
  155. static struct connection *nodeid2con(int nodeid, gfp_t alloc)
  156. {
  157. struct connection *con, *tmp;
  158. int r;
  159. con = __find_con(nodeid);
  160. if (con || !alloc)
  161. return con;
  162. con = kzalloc(sizeof(*con), alloc);
  163. if (!con)
  164. return NULL;
  165. con->rx_buflen = dlm_config.ci_buffer_size;
  166. con->rx_buf = kmalloc(con->rx_buflen, GFP_NOFS);
  167. if (!con->rx_buf) {
  168. kfree(con);
  169. return NULL;
  170. }
  171. con->nodeid = nodeid;
  172. mutex_init(&con->sock_mutex);
  173. INIT_LIST_HEAD(&con->writequeue);
  174. spin_lock_init(&con->writequeue_lock);
  175. INIT_WORK(&con->swork, process_send_sockets);
  176. INIT_WORK(&con->rwork, process_recv_sockets);
  177. init_waitqueue_head(&con->shutdown_wait);
  178. /* Setup action pointers for child sockets */
  179. if (con->nodeid) {
  180. struct connection *zerocon = __find_con(0);
  181. con->connect_action = zerocon->connect_action;
  182. if (!con->rx_action)
  183. con->rx_action = zerocon->rx_action;
  184. }
  185. r = nodeid_hash(nodeid);
  186. spin_lock(&connections_lock);
  187. /* Because multiple workqueues/threads calls this function it can
  188. * race on multiple cpu's. Instead of locking hot path __find_con()
  189. * we just check in rare cases of recently added nodes again
  190. * under protection of connections_lock. If this is the case we
  191. * abort our connection creation and return the existing connection.
  192. */
  193. tmp = __find_con(nodeid);
  194. if (tmp) {
  195. spin_unlock(&connections_lock);
  196. kfree(con->rx_buf);
  197. kfree(con);
  198. return tmp;
  199. }
  200. hlist_add_head_rcu(&con->list, &connection_hash[r]);
  201. spin_unlock(&connections_lock);
  202. return con;
  203. }
  204. /* Loop round all connections */
  205. static void foreach_conn(void (*conn_func)(struct connection *c))
  206. {
  207. int i, idx;
  208. struct connection *con;
  209. idx = srcu_read_lock(&connections_srcu);
  210. for (i = 0; i < CONN_HASH_SIZE; i++) {
  211. hlist_for_each_entry_rcu(con, &connection_hash[i], list)
  212. conn_func(con);
  213. }
  214. srcu_read_unlock(&connections_srcu, idx);
  215. }
  216. static struct dlm_node_addr *find_node_addr(int nodeid)
  217. {
  218. struct dlm_node_addr *na;
  219. list_for_each_entry(na, &dlm_node_addrs, list) {
  220. if (na->nodeid == nodeid)
  221. return na;
  222. }
  223. return NULL;
  224. }
  225. static int addr_compare(struct sockaddr_storage *x, struct sockaddr_storage *y)
  226. {
  227. switch (x->ss_family) {
  228. case AF_INET: {
  229. struct sockaddr_in *sinx = (struct sockaddr_in *)x;
  230. struct sockaddr_in *siny = (struct sockaddr_in *)y;
  231. if (sinx->sin_addr.s_addr != siny->sin_addr.s_addr)
  232. return 0;
  233. if (sinx->sin_port != siny->sin_port)
  234. return 0;
  235. break;
  236. }
  237. case AF_INET6: {
  238. struct sockaddr_in6 *sinx = (struct sockaddr_in6 *)x;
  239. struct sockaddr_in6 *siny = (struct sockaddr_in6 *)y;
  240. if (!ipv6_addr_equal(&sinx->sin6_addr, &siny->sin6_addr))
  241. return 0;
  242. if (sinx->sin6_port != siny->sin6_port)
  243. return 0;
  244. break;
  245. }
  246. default:
  247. return 0;
  248. }
  249. return 1;
  250. }
  251. static int nodeid_to_addr(int nodeid, struct sockaddr_storage *sas_out,
  252. struct sockaddr *sa_out, bool try_new_addr)
  253. {
  254. struct sockaddr_storage sas;
  255. struct dlm_node_addr *na;
  256. if (!dlm_local_count)
  257. return -1;
  258. spin_lock(&dlm_node_addrs_spin);
  259. na = find_node_addr(nodeid);
  260. if (na && na->addr_count) {
  261. memcpy(&sas, na->addr[na->curr_addr_index],
  262. sizeof(struct sockaddr_storage));
  263. if (try_new_addr) {
  264. na->curr_addr_index++;
  265. if (na->curr_addr_index == na->addr_count)
  266. na->curr_addr_index = 0;
  267. }
  268. }
  269. spin_unlock(&dlm_node_addrs_spin);
  270. if (!na)
  271. return -EEXIST;
  272. if (!na->addr_count)
  273. return -ENOENT;
  274. if (sas_out)
  275. memcpy(sas_out, &sas, sizeof(struct sockaddr_storage));
  276. if (!sa_out)
  277. return 0;
  278. if (dlm_local_addr[0]->ss_family == AF_INET) {
  279. struct sockaddr_in *in4 = (struct sockaddr_in *) &sas;
  280. struct sockaddr_in *ret4 = (struct sockaddr_in *) sa_out;
  281. ret4->sin_addr.s_addr = in4->sin_addr.s_addr;
  282. } else {
  283. struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) &sas;
  284. struct sockaddr_in6 *ret6 = (struct sockaddr_in6 *) sa_out;
  285. ret6->sin6_addr = in6->sin6_addr;
  286. }
  287. return 0;
  288. }
  289. static int addr_to_nodeid(struct sockaddr_storage *addr, int *nodeid)
  290. {
  291. struct dlm_node_addr *na;
  292. int rv = -EEXIST;
  293. int addr_i;
  294. spin_lock(&dlm_node_addrs_spin);
  295. list_for_each_entry(na, &dlm_node_addrs, list) {
  296. if (!na->addr_count)
  297. continue;
  298. for (addr_i = 0; addr_i < na->addr_count; addr_i++) {
  299. if (addr_compare(na->addr[addr_i], addr)) {
  300. *nodeid = na->nodeid;
  301. rv = 0;
  302. goto unlock;
  303. }
  304. }
  305. }
  306. unlock:
  307. spin_unlock(&dlm_node_addrs_spin);
  308. return rv;
  309. }
  310. int dlm_lowcomms_addr(int nodeid, struct sockaddr_storage *addr, int len)
  311. {
  312. struct sockaddr_storage *new_addr;
  313. struct dlm_node_addr *new_node, *na;
  314. new_node = kzalloc(sizeof(struct dlm_node_addr), GFP_NOFS);
  315. if (!new_node)
  316. return -ENOMEM;
  317. new_addr = kzalloc(sizeof(struct sockaddr_storage), GFP_NOFS);
  318. if (!new_addr) {
  319. kfree(new_node);
  320. return -ENOMEM;
  321. }
  322. memcpy(new_addr, addr, len);
  323. spin_lock(&dlm_node_addrs_spin);
  324. na = find_node_addr(nodeid);
  325. if (!na) {
  326. new_node->nodeid = nodeid;
  327. new_node->addr[0] = new_addr;
  328. new_node->addr_count = 1;
  329. list_add(&new_node->list, &dlm_node_addrs);
  330. spin_unlock(&dlm_node_addrs_spin);
  331. return 0;
  332. }
  333. if (na->addr_count >= DLM_MAX_ADDR_COUNT) {
  334. spin_unlock(&dlm_node_addrs_spin);
  335. kfree(new_addr);
  336. kfree(new_node);
  337. return -ENOSPC;
  338. }
  339. na->addr[na->addr_count++] = new_addr;
  340. spin_unlock(&dlm_node_addrs_spin);
  341. kfree(new_node);
  342. return 0;
  343. }
  344. /* Data available on socket or listen socket received a connect */
  345. static void lowcomms_data_ready(struct sock *sk)
  346. {
  347. struct connection *con;
  348. read_lock_bh(&sk->sk_callback_lock);
  349. con = sock2con(sk);
  350. if (con && !test_and_set_bit(CF_READ_PENDING, &con->flags))
  351. queue_work(recv_workqueue, &con->rwork);
  352. read_unlock_bh(&sk->sk_callback_lock);
  353. }
  354. static void lowcomms_write_space(struct sock *sk)
  355. {
  356. struct connection *con;
  357. read_lock_bh(&sk->sk_callback_lock);
  358. con = sock2con(sk);
  359. if (!con)
  360. goto out;
  361. clear_bit(SOCK_NOSPACE, &con->sock->flags);
  362. if (test_and_clear_bit(CF_APP_LIMITED, &con->flags)) {
  363. con->sock->sk->sk_write_pending--;
  364. clear_bit(SOCKWQ_ASYNC_NOSPACE, &con->sock->flags);
  365. }
  366. queue_work(send_workqueue, &con->swork);
  367. out:
  368. read_unlock_bh(&sk->sk_callback_lock);
  369. }
  370. static inline void lowcomms_connect_sock(struct connection *con)
  371. {
  372. if (test_bit(CF_CLOSE, &con->flags))
  373. return;
  374. queue_work(send_workqueue, &con->swork);
  375. cond_resched();
  376. }
  377. static void lowcomms_state_change(struct sock *sk)
  378. {
  379. /* SCTP layer is not calling sk_data_ready when the connection
  380. * is done, so we catch the signal through here. Also, it
  381. * doesn't switch socket state when entering shutdown, so we
  382. * skip the write in that case.
  383. */
  384. if (sk->sk_shutdown) {
  385. if (sk->sk_shutdown == RCV_SHUTDOWN)
  386. lowcomms_data_ready(sk);
  387. } else if (sk->sk_state == TCP_ESTABLISHED) {
  388. lowcomms_write_space(sk);
  389. }
  390. }
  391. int dlm_lowcomms_connect_node(int nodeid)
  392. {
  393. struct connection *con;
  394. if (nodeid == dlm_our_nodeid())
  395. return 0;
  396. con = nodeid2con(nodeid, GFP_NOFS);
  397. if (!con)
  398. return -ENOMEM;
  399. lowcomms_connect_sock(con);
  400. return 0;
  401. }
  402. static void lowcomms_error_report(struct sock *sk)
  403. {
  404. struct connection *con;
  405. void (*orig_report)(struct sock *) = NULL;
  406. struct inet_sock *inet;
  407. read_lock_bh(&sk->sk_callback_lock);
  408. con = sock2con(sk);
  409. if (con == NULL)
  410. goto out;
  411. orig_report = listen_sock.sk_error_report;
  412. inet = inet_sk(sk);
  413. switch (sk->sk_family) {
  414. case AF_INET:
  415. printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
  416. "sending to node %d at %pI4, dport %d, "
  417. "sk_err=%d/%d\n", dlm_our_nodeid(),
  418. con->nodeid, &inet->inet_daddr,
  419. ntohs(inet->inet_dport), sk->sk_err,
  420. sk->sk_err_soft);
  421. break;
  422. #if IS_ENABLED(CONFIG_IPV6)
  423. case AF_INET6:
  424. printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
  425. "sending to node %d at %pI6c, "
  426. "dport %d, sk_err=%d/%d\n", dlm_our_nodeid(),
  427. con->nodeid, &sk->sk_v6_daddr,
  428. ntohs(inet->inet_dport), sk->sk_err,
  429. sk->sk_err_soft);
  430. break;
  431. #endif
  432. default:
  433. printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
  434. "invalid socket family %d set, "
  435. "sk_err=%d/%d\n", dlm_our_nodeid(),
  436. sk->sk_family, sk->sk_err, sk->sk_err_soft);
  437. goto out;
  438. }
  439. out:
  440. read_unlock_bh(&sk->sk_callback_lock);
  441. if (orig_report)
  442. orig_report(sk);
  443. }
  444. /* Note: sk_callback_lock must be locked before calling this function. */
  445. static void save_listen_callbacks(struct socket *sock)
  446. {
  447. struct sock *sk = sock->sk;
  448. listen_sock.sk_data_ready = sk->sk_data_ready;
  449. listen_sock.sk_state_change = sk->sk_state_change;
  450. listen_sock.sk_write_space = sk->sk_write_space;
  451. listen_sock.sk_error_report = sk->sk_error_report;
  452. }
  453. static void restore_callbacks(struct socket *sock)
  454. {
  455. struct sock *sk = sock->sk;
  456. write_lock_bh(&sk->sk_callback_lock);
  457. sk->sk_user_data = NULL;
  458. sk->sk_data_ready = listen_sock.sk_data_ready;
  459. sk->sk_state_change = listen_sock.sk_state_change;
  460. sk->sk_write_space = listen_sock.sk_write_space;
  461. sk->sk_error_report = listen_sock.sk_error_report;
  462. write_unlock_bh(&sk->sk_callback_lock);
  463. }
  464. /* Make a socket active */
  465. static void add_sock(struct socket *sock, struct connection *con)
  466. {
  467. struct sock *sk = sock->sk;
  468. write_lock_bh(&sk->sk_callback_lock);
  469. con->sock = sock;
  470. sk->sk_user_data = con;
  471. /* Install a data_ready callback */
  472. sk->sk_data_ready = lowcomms_data_ready;
  473. sk->sk_write_space = lowcomms_write_space;
  474. sk->sk_state_change = lowcomms_state_change;
  475. sk->sk_allocation = GFP_NOFS;
  476. sk->sk_error_report = lowcomms_error_report;
  477. write_unlock_bh(&sk->sk_callback_lock);
  478. }
  479. /* Add the port number to an IPv6 or 4 sockaddr and return the address
  480. length */
  481. static void make_sockaddr(struct sockaddr_storage *saddr, uint16_t port,
  482. int *addr_len)
  483. {
  484. saddr->ss_family = dlm_local_addr[0]->ss_family;
  485. if (saddr->ss_family == AF_INET) {
  486. struct sockaddr_in *in4_addr = (struct sockaddr_in *)saddr;
  487. in4_addr->sin_port = cpu_to_be16(port);
  488. *addr_len = sizeof(struct sockaddr_in);
  489. memset(&in4_addr->sin_zero, 0, sizeof(in4_addr->sin_zero));
  490. } else {
  491. struct sockaddr_in6 *in6_addr = (struct sockaddr_in6 *)saddr;
  492. in6_addr->sin6_port = cpu_to_be16(port);
  493. *addr_len = sizeof(struct sockaddr_in6);
  494. }
  495. memset((char *)saddr + *addr_len, 0, sizeof(struct sockaddr_storage) - *addr_len);
  496. }
  497. /* Close a remote connection and tidy up */
  498. static void close_connection(struct connection *con, bool and_other,
  499. bool tx, bool rx)
  500. {
  501. bool closing = test_and_set_bit(CF_CLOSING, &con->flags);
  502. if (tx && !closing && cancel_work_sync(&con->swork)) {
  503. log_print("canceled swork for node %d", con->nodeid);
  504. clear_bit(CF_WRITE_PENDING, &con->flags);
  505. }
  506. if (rx && !closing && cancel_work_sync(&con->rwork)) {
  507. log_print("canceled rwork for node %d", con->nodeid);
  508. clear_bit(CF_READ_PENDING, &con->flags);
  509. }
  510. mutex_lock(&con->sock_mutex);
  511. if (con->sock) {
  512. restore_callbacks(con->sock);
  513. sock_release(con->sock);
  514. con->sock = NULL;
  515. }
  516. if (con->othercon && and_other) {
  517. /* Will only re-enter once. */
  518. close_connection(con->othercon, false, tx, rx);
  519. }
  520. con->rx_leftover = 0;
  521. con->retries = 0;
  522. mutex_unlock(&con->sock_mutex);
  523. clear_bit(CF_CLOSING, &con->flags);
  524. }
  525. static void shutdown_connection(struct connection *con)
  526. {
  527. int ret;
  528. flush_work(&con->swork);
  529. mutex_lock(&con->sock_mutex);
  530. /* nothing to shutdown */
  531. if (!con->sock) {
  532. mutex_unlock(&con->sock_mutex);
  533. return;
  534. }
  535. set_bit(CF_SHUTDOWN, &con->flags);
  536. ret = kernel_sock_shutdown(con->sock, SHUT_WR);
  537. mutex_unlock(&con->sock_mutex);
  538. if (ret) {
  539. log_print("Connection %p failed to shutdown: %d will force close",
  540. con, ret);
  541. goto force_close;
  542. } else {
  543. ret = wait_event_timeout(con->shutdown_wait,
  544. !test_bit(CF_SHUTDOWN, &con->flags),
  545. DLM_SHUTDOWN_WAIT_TIMEOUT);
  546. if (ret == 0) {
  547. log_print("Connection %p shutdown timed out, will force close",
  548. con);
  549. goto force_close;
  550. }
  551. }
  552. return;
  553. force_close:
  554. clear_bit(CF_SHUTDOWN, &con->flags);
  555. close_connection(con, false, true, true);
  556. }
  557. static void dlm_tcp_shutdown(struct connection *con)
  558. {
  559. if (con->othercon)
  560. shutdown_connection(con->othercon);
  561. shutdown_connection(con);
  562. }
  563. static int con_realloc_receive_buf(struct connection *con, int newlen)
  564. {
  565. unsigned char *newbuf;
  566. newbuf = kmalloc(newlen, GFP_NOFS);
  567. if (!newbuf)
  568. return -ENOMEM;
  569. /* copy any leftover from last receive */
  570. if (con->rx_leftover)
  571. memmove(newbuf, con->rx_buf, con->rx_leftover);
  572. /* swap to new buffer space */
  573. kfree(con->rx_buf);
  574. con->rx_buflen = newlen;
  575. con->rx_buf = newbuf;
  576. return 0;
  577. }
  578. /* Data received from remote end */
  579. static int receive_from_sock(struct connection *con)
  580. {
  581. int call_again_soon = 0;
  582. struct msghdr msg;
  583. struct kvec iov;
  584. int ret, buflen;
  585. mutex_lock(&con->sock_mutex);
  586. if (con->sock == NULL) {
  587. ret = -EAGAIN;
  588. goto out_close;
  589. }
  590. if (con->nodeid == 0) {
  591. ret = -EINVAL;
  592. goto out_close;
  593. }
  594. /* realloc if we get new buffer size to read out */
  595. buflen = dlm_config.ci_buffer_size;
  596. if (con->rx_buflen != buflen && con->rx_leftover <= buflen) {
  597. ret = con_realloc_receive_buf(con, buflen);
  598. if (ret < 0)
  599. goto out_resched;
  600. }
  601. /* calculate new buffer parameter regarding last receive and
  602. * possible leftover bytes
  603. */
  604. iov.iov_base = con->rx_buf + con->rx_leftover;
  605. iov.iov_len = con->rx_buflen - con->rx_leftover;
  606. memset(&msg, 0, sizeof(msg));
  607. msg.msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL;
  608. ret = kernel_recvmsg(con->sock, &msg, &iov, 1, iov.iov_len,
  609. msg.msg_flags);
  610. if (ret <= 0)
  611. goto out_close;
  612. else if (ret == iov.iov_len)
  613. call_again_soon = 1;
  614. /* new buflen according readed bytes and leftover from last receive */
  615. buflen = ret + con->rx_leftover;
  616. ret = dlm_process_incoming_buffer(con->nodeid, con->rx_buf, buflen);
  617. if (ret < 0)
  618. goto out_close;
  619. /* calculate leftover bytes from process and put it into begin of
  620. * the receive buffer, so next receive we have the full message
  621. * at the start address of the receive buffer.
  622. */
  623. con->rx_leftover = buflen - ret;
  624. if (con->rx_leftover) {
  625. memmove(con->rx_buf, con->rx_buf + ret,
  626. con->rx_leftover);
  627. call_again_soon = true;
  628. }
  629. if (call_again_soon)
  630. goto out_resched;
  631. mutex_unlock(&con->sock_mutex);
  632. return 0;
  633. out_resched:
  634. if (!test_and_set_bit(CF_READ_PENDING, &con->flags))
  635. queue_work(recv_workqueue, &con->rwork);
  636. mutex_unlock(&con->sock_mutex);
  637. return -EAGAIN;
  638. out_close:
  639. mutex_unlock(&con->sock_mutex);
  640. if (ret != -EAGAIN) {
  641. /* Reconnect when there is something to send */
  642. close_connection(con, false, true, false);
  643. if (ret == 0) {
  644. log_print("connection %p got EOF from %d",
  645. con, con->nodeid);
  646. /* handling for tcp shutdown */
  647. clear_bit(CF_SHUTDOWN, &con->flags);
  648. wake_up(&con->shutdown_wait);
  649. /* signal to breaking receive worker */
  650. ret = -1;
  651. }
  652. }
  653. return ret;
  654. }
  655. /* Listening socket is busy, accept a connection */
  656. static int accept_from_sock(struct connection *con)
  657. {
  658. int result;
  659. struct sockaddr_storage peeraddr;
  660. struct socket *newsock;
  661. int len;
  662. int nodeid;
  663. struct connection *newcon;
  664. struct connection *addcon;
  665. unsigned int mark;
  666. if (!dlm_allow_conn) {
  667. return -1;
  668. }
  669. mutex_lock_nested(&con->sock_mutex, 0);
  670. if (!con->sock) {
  671. mutex_unlock(&con->sock_mutex);
  672. return -ENOTCONN;
  673. }
  674. result = kernel_accept(con->sock, &newsock, O_NONBLOCK);
  675. if (result < 0)
  676. goto accept_err;
  677. /* Get the connected socket's peer */
  678. memset(&peeraddr, 0, sizeof(peeraddr));
  679. len = newsock->ops->getname(newsock, (struct sockaddr *)&peeraddr, 2);
  680. if (len < 0) {
  681. result = -ECONNABORTED;
  682. goto accept_err;
  683. }
  684. /* Get the new node's NODEID */
  685. make_sockaddr(&peeraddr, 0, &len);
  686. if (addr_to_nodeid(&peeraddr, &nodeid)) {
  687. unsigned char *b=(unsigned char *)&peeraddr;
  688. log_print("connect from non cluster node");
  689. print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE,
  690. b, sizeof(struct sockaddr_storage));
  691. sock_release(newsock);
  692. mutex_unlock(&con->sock_mutex);
  693. return -1;
  694. }
  695. dlm_comm_mark(nodeid, &mark);
  696. sock_set_mark(newsock->sk, mark);
  697. log_print("got connection from %d", nodeid);
  698. /* Check to see if we already have a connection to this node. This
  699. * could happen if the two nodes initiate a connection at roughly
  700. * the same time and the connections cross on the wire.
  701. * In this case we store the incoming one in "othercon"
  702. */
  703. newcon = nodeid2con(nodeid, GFP_NOFS);
  704. if (!newcon) {
  705. result = -ENOMEM;
  706. goto accept_err;
  707. }
  708. mutex_lock_nested(&newcon->sock_mutex, 1);
  709. if (newcon->sock) {
  710. struct connection *othercon = newcon->othercon;
  711. if (!othercon) {
  712. othercon = kzalloc(sizeof(*othercon), GFP_NOFS);
  713. if (!othercon) {
  714. log_print("failed to allocate incoming socket");
  715. mutex_unlock(&newcon->sock_mutex);
  716. result = -ENOMEM;
  717. goto accept_err;
  718. }
  719. othercon->rx_buflen = dlm_config.ci_buffer_size;
  720. othercon->rx_buf = kmalloc(othercon->rx_buflen, GFP_NOFS);
  721. if (!othercon->rx_buf) {
  722. mutex_unlock(&newcon->sock_mutex);
  723. kfree(othercon);
  724. log_print("failed to allocate incoming socket receive buffer");
  725. result = -ENOMEM;
  726. goto accept_err;
  727. }
  728. othercon->nodeid = nodeid;
  729. othercon->rx_action = receive_from_sock;
  730. mutex_init(&othercon->sock_mutex);
  731. INIT_LIST_HEAD(&othercon->writequeue);
  732. spin_lock_init(&othercon->writequeue_lock);
  733. INIT_WORK(&othercon->swork, process_send_sockets);
  734. INIT_WORK(&othercon->rwork, process_recv_sockets);
  735. init_waitqueue_head(&othercon->shutdown_wait);
  736. set_bit(CF_IS_OTHERCON, &othercon->flags);
  737. } else {
  738. /* close other sock con if we have something new */
  739. close_connection(othercon, false, true, false);
  740. }
  741. mutex_lock_nested(&othercon->sock_mutex, 2);
  742. newcon->othercon = othercon;
  743. add_sock(newsock, othercon);
  744. addcon = othercon;
  745. mutex_unlock(&othercon->sock_mutex);
  746. }
  747. else {
  748. newcon->rx_action = receive_from_sock;
  749. /* accept copies the sk after we've saved the callbacks, so we
  750. don't want to save them a second time or comm errors will
  751. result in calling sk_error_report recursively. */
  752. add_sock(newsock, newcon);
  753. addcon = newcon;
  754. }
  755. mutex_unlock(&newcon->sock_mutex);
  756. /*
  757. * Add it to the active queue in case we got data
  758. * between processing the accept adding the socket
  759. * to the read_sockets list
  760. */
  761. if (!test_and_set_bit(CF_READ_PENDING, &addcon->flags))
  762. queue_work(recv_workqueue, &addcon->rwork);
  763. mutex_unlock(&con->sock_mutex);
  764. return 0;
  765. accept_err:
  766. mutex_unlock(&con->sock_mutex);
  767. if (newsock)
  768. sock_release(newsock);
  769. if (result != -EAGAIN)
  770. log_print("error accepting connection from node: %d", result);
  771. return result;
  772. }
  773. static void free_entry(struct writequeue_entry *e)
  774. {
  775. __free_page(e->page);
  776. kfree(e);
  777. }
  778. /*
  779. * writequeue_entry_complete - try to delete and free write queue entry
  780. * @e: write queue entry to try to delete
  781. * @completed: bytes completed
  782. *
  783. * writequeue_lock must be held.
  784. */
  785. static void writequeue_entry_complete(struct writequeue_entry *e, int completed)
  786. {
  787. e->offset += completed;
  788. e->len -= completed;
  789. if (e->len == 0 && e->users == 0) {
  790. list_del(&e->list);
  791. free_entry(e);
  792. }
  793. }
  794. /*
  795. * sctp_bind_addrs - bind a SCTP socket to all our addresses
  796. */
  797. static int sctp_bind_addrs(struct connection *con, uint16_t port)
  798. {
  799. struct sockaddr_storage localaddr;
  800. struct sockaddr *addr = (struct sockaddr *)&localaddr;
  801. int i, addr_len, result = 0;
  802. for (i = 0; i < dlm_local_count; i++) {
  803. memcpy(&localaddr, dlm_local_addr[i], sizeof(localaddr));
  804. make_sockaddr(&localaddr, port, &addr_len);
  805. if (!i)
  806. result = kernel_bind(con->sock, addr, addr_len);
  807. else
  808. result = sock_bind_add(con->sock->sk, addr, addr_len);
  809. if (result < 0) {
  810. log_print("Can't bind to %d addr number %d, %d.\n",
  811. port, i + 1, result);
  812. break;
  813. }
  814. }
  815. return result;
  816. }
  817. /* Initiate an SCTP association.
  818. This is a special case of send_to_sock() in that we don't yet have a
  819. peeled-off socket for this association, so we use the listening socket
  820. and add the primary IP address of the remote node.
  821. */
  822. static void sctp_connect_to_sock(struct connection *con)
  823. {
  824. struct sockaddr_storage daddr;
  825. int result;
  826. int addr_len;
  827. struct socket *sock;
  828. unsigned int mark;
  829. if (con->nodeid == 0) {
  830. log_print("attempt to connect sock 0 foiled");
  831. return;
  832. }
  833. dlm_comm_mark(con->nodeid, &mark);
  834. mutex_lock(&con->sock_mutex);
  835. /* Some odd races can cause double-connects, ignore them */
  836. if (con->retries++ > MAX_CONNECT_RETRIES)
  837. goto out;
  838. if (con->sock) {
  839. log_print("node %d already connected.", con->nodeid);
  840. goto out;
  841. }
  842. memset(&daddr, 0, sizeof(daddr));
  843. result = nodeid_to_addr(con->nodeid, &daddr, NULL, true);
  844. if (result < 0) {
  845. log_print("no address for nodeid %d", con->nodeid);
  846. goto out;
  847. }
  848. /* Create a socket to communicate with */
  849. result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
  850. SOCK_STREAM, IPPROTO_SCTP, &sock);
  851. if (result < 0)
  852. goto socket_err;
  853. sock_set_mark(sock->sk, mark);
  854. con->rx_action = receive_from_sock;
  855. con->connect_action = sctp_connect_to_sock;
  856. add_sock(sock, con);
  857. /* Bind to all addresses. */
  858. if (sctp_bind_addrs(con, 0))
  859. goto bind_err;
  860. make_sockaddr(&daddr, dlm_config.ci_tcp_port, &addr_len);
  861. log_print("connecting to %d", con->nodeid);
  862. /* Turn off Nagle's algorithm */
  863. sctp_sock_set_nodelay(sock->sk);
  864. /*
  865. * Make sock->ops->connect() function return in specified time,
  866. * since O_NONBLOCK argument in connect() function does not work here,
  867. * then, we should restore the default value of this attribute.
  868. */
  869. sock_set_sndtimeo(sock->sk, 5);
  870. result = sock->ops->connect(sock, (struct sockaddr *)&daddr, addr_len,
  871. 0);
  872. sock_set_sndtimeo(sock->sk, 0);
  873. if (result == -EINPROGRESS)
  874. result = 0;
  875. if (result == 0)
  876. goto out;
  877. bind_err:
  878. con->sock = NULL;
  879. sock_release(sock);
  880. socket_err:
  881. /*
  882. * Some errors are fatal and this list might need adjusting. For other
  883. * errors we try again until the max number of retries is reached.
  884. */
  885. if (result != -EHOSTUNREACH &&
  886. result != -ENETUNREACH &&
  887. result != -ENETDOWN &&
  888. result != -EINVAL &&
  889. result != -EPROTONOSUPPORT) {
  890. log_print("connect %d try %d error %d", con->nodeid,
  891. con->retries, result);
  892. mutex_unlock(&con->sock_mutex);
  893. msleep(1000);
  894. lowcomms_connect_sock(con);
  895. return;
  896. }
  897. out:
  898. mutex_unlock(&con->sock_mutex);
  899. }
  900. /* Connect a new socket to its peer */
  901. static void tcp_connect_to_sock(struct connection *con)
  902. {
  903. struct sockaddr_storage saddr, src_addr;
  904. int addr_len;
  905. struct socket *sock = NULL;
  906. unsigned int mark;
  907. int result;
  908. if (con->nodeid == 0) {
  909. log_print("attempt to connect sock 0 foiled");
  910. return;
  911. }
  912. dlm_comm_mark(con->nodeid, &mark);
  913. mutex_lock(&con->sock_mutex);
  914. if (con->retries++ > MAX_CONNECT_RETRIES)
  915. goto out;
  916. /* Some odd races can cause double-connects, ignore them */
  917. if (con->sock)
  918. goto out;
  919. /* Create a socket to communicate with */
  920. result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
  921. SOCK_STREAM, IPPROTO_TCP, &sock);
  922. if (result < 0)
  923. goto out_err;
  924. sock_set_mark(sock->sk, mark);
  925. memset(&saddr, 0, sizeof(saddr));
  926. result = nodeid_to_addr(con->nodeid, &saddr, NULL, false);
  927. if (result < 0) {
  928. log_print("no address for nodeid %d", con->nodeid);
  929. goto out_err;
  930. }
  931. con->rx_action = receive_from_sock;
  932. con->connect_action = tcp_connect_to_sock;
  933. con->shutdown_action = dlm_tcp_shutdown;
  934. add_sock(sock, con);
  935. /* Bind to our cluster-known address connecting to avoid
  936. routing problems */
  937. memcpy(&src_addr, dlm_local_addr[0], sizeof(src_addr));
  938. make_sockaddr(&src_addr, 0, &addr_len);
  939. result = sock->ops->bind(sock, (struct sockaddr *) &src_addr,
  940. addr_len);
  941. if (result < 0) {
  942. log_print("could not bind for connect: %d", result);
  943. /* This *may* not indicate a critical error */
  944. }
  945. make_sockaddr(&saddr, dlm_config.ci_tcp_port, &addr_len);
  946. log_print("connecting to %d", con->nodeid);
  947. /* Turn off Nagle's algorithm */
  948. tcp_sock_set_nodelay(sock->sk);
  949. result = sock->ops->connect(sock, (struct sockaddr *)&saddr, addr_len,
  950. O_NONBLOCK);
  951. if (result == -EINPROGRESS)
  952. result = 0;
  953. if (result == 0)
  954. goto out;
  955. out_err:
  956. if (con->sock) {
  957. sock_release(con->sock);
  958. con->sock = NULL;
  959. } else if (sock) {
  960. sock_release(sock);
  961. }
  962. /*
  963. * Some errors are fatal and this list might need adjusting. For other
  964. * errors we try again until the max number of retries is reached.
  965. */
  966. if (result != -EHOSTUNREACH &&
  967. result != -ENETUNREACH &&
  968. result != -ENETDOWN &&
  969. result != -EINVAL &&
  970. result != -EPROTONOSUPPORT) {
  971. log_print("connect %d try %d error %d", con->nodeid,
  972. con->retries, result);
  973. mutex_unlock(&con->sock_mutex);
  974. msleep(1000);
  975. lowcomms_connect_sock(con);
  976. return;
  977. }
  978. out:
  979. mutex_unlock(&con->sock_mutex);
  980. return;
  981. }
  982. static struct socket *tcp_create_listen_sock(struct connection *con,
  983. struct sockaddr_storage *saddr)
  984. {
  985. struct socket *sock = NULL;
  986. int result = 0;
  987. int addr_len;
  988. if (dlm_local_addr[0]->ss_family == AF_INET)
  989. addr_len = sizeof(struct sockaddr_in);
  990. else
  991. addr_len = sizeof(struct sockaddr_in6);
  992. /* Create a socket to communicate with */
  993. result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
  994. SOCK_STREAM, IPPROTO_TCP, &sock);
  995. if (result < 0) {
  996. log_print("Can't create listening comms socket");
  997. goto create_out;
  998. }
  999. sock_set_mark(sock->sk, dlm_config.ci_mark);
  1000. /* Turn off Nagle's algorithm */
  1001. tcp_sock_set_nodelay(sock->sk);
  1002. sock_set_reuseaddr(sock->sk);
  1003. write_lock_bh(&sock->sk->sk_callback_lock);
  1004. sock->sk->sk_user_data = con;
  1005. save_listen_callbacks(sock);
  1006. con->rx_action = accept_from_sock;
  1007. con->connect_action = tcp_connect_to_sock;
  1008. write_unlock_bh(&sock->sk->sk_callback_lock);
  1009. /* Bind to our port */
  1010. make_sockaddr(saddr, dlm_config.ci_tcp_port, &addr_len);
  1011. result = sock->ops->bind(sock, (struct sockaddr *) saddr, addr_len);
  1012. if (result < 0) {
  1013. log_print("Can't bind to port %d", dlm_config.ci_tcp_port);
  1014. sock_release(sock);
  1015. sock = NULL;
  1016. con->sock = NULL;
  1017. goto create_out;
  1018. }
  1019. sock_set_keepalive(sock->sk);
  1020. result = sock->ops->listen(sock, 5);
  1021. if (result < 0) {
  1022. log_print("Can't listen on port %d", dlm_config.ci_tcp_port);
  1023. sock_release(sock);
  1024. sock = NULL;
  1025. goto create_out;
  1026. }
  1027. create_out:
  1028. return sock;
  1029. }
  1030. /* Get local addresses */
  1031. static void init_local(void)
  1032. {
  1033. struct sockaddr_storage sas, *addr;
  1034. int i;
  1035. dlm_local_count = 0;
  1036. for (i = 0; i < DLM_MAX_ADDR_COUNT; i++) {
  1037. if (dlm_our_addr(&sas, i))
  1038. break;
  1039. addr = kmemdup(&sas, sizeof(*addr), GFP_NOFS);
  1040. if (!addr)
  1041. break;
  1042. dlm_local_addr[dlm_local_count++] = addr;
  1043. }
  1044. }
  1045. static void deinit_local(void)
  1046. {
  1047. int i;
  1048. for (i = 0; i < dlm_local_count; i++)
  1049. kfree(dlm_local_addr[i]);
  1050. }
  1051. /* Initialise SCTP socket and bind to all interfaces */
  1052. static int sctp_listen_for_all(void)
  1053. {
  1054. struct socket *sock = NULL;
  1055. int result = -EINVAL;
  1056. struct connection *con = nodeid2con(0, GFP_NOFS);
  1057. if (!con)
  1058. return -ENOMEM;
  1059. log_print("Using SCTP for communications");
  1060. result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
  1061. SOCK_STREAM, IPPROTO_SCTP, &sock);
  1062. if (result < 0) {
  1063. log_print("Can't create comms socket, check SCTP is loaded");
  1064. goto out;
  1065. }
  1066. sock_set_rcvbuf(sock->sk, NEEDED_RMEM);
  1067. sock_set_mark(sock->sk, dlm_config.ci_mark);
  1068. sctp_sock_set_nodelay(sock->sk);
  1069. write_lock_bh(&sock->sk->sk_callback_lock);
  1070. /* Init con struct */
  1071. sock->sk->sk_user_data = con;
  1072. save_listen_callbacks(sock);
  1073. con->sock = sock;
  1074. con->sock->sk->sk_data_ready = lowcomms_data_ready;
  1075. con->rx_action = accept_from_sock;
  1076. con->connect_action = sctp_connect_to_sock;
  1077. write_unlock_bh(&sock->sk->sk_callback_lock);
  1078. /* Bind to all addresses. */
  1079. if (sctp_bind_addrs(con, dlm_config.ci_tcp_port))
  1080. goto create_delsock;
  1081. result = sock->ops->listen(sock, 5);
  1082. if (result < 0) {
  1083. log_print("Can't set socket listening");
  1084. goto create_delsock;
  1085. }
  1086. return 0;
  1087. create_delsock:
  1088. sock_release(sock);
  1089. con->sock = NULL;
  1090. out:
  1091. return result;
  1092. }
  1093. static int tcp_listen_for_all(void)
  1094. {
  1095. struct socket *sock = NULL;
  1096. struct connection *con = nodeid2con(0, GFP_NOFS);
  1097. int result = -EINVAL;
  1098. if (!con)
  1099. return -ENOMEM;
  1100. /* We don't support multi-homed hosts */
  1101. if (dlm_local_addr[1] != NULL) {
  1102. log_print("TCP protocol can't handle multi-homed hosts, "
  1103. "try SCTP");
  1104. return -EINVAL;
  1105. }
  1106. log_print("Using TCP for communications");
  1107. sock = tcp_create_listen_sock(con, dlm_local_addr[0]);
  1108. if (sock) {
  1109. add_sock(sock, con);
  1110. result = 0;
  1111. }
  1112. else {
  1113. result = -EADDRINUSE;
  1114. }
  1115. return result;
  1116. }
  1117. static struct writequeue_entry *new_writequeue_entry(struct connection *con,
  1118. gfp_t allocation)
  1119. {
  1120. struct writequeue_entry *entry;
  1121. entry = kmalloc(sizeof(struct writequeue_entry), allocation);
  1122. if (!entry)
  1123. return NULL;
  1124. entry->page = alloc_page(allocation);
  1125. if (!entry->page) {
  1126. kfree(entry);
  1127. return NULL;
  1128. }
  1129. entry->offset = 0;
  1130. entry->len = 0;
  1131. entry->end = 0;
  1132. entry->users = 0;
  1133. entry->con = con;
  1134. return entry;
  1135. }
  1136. void *dlm_lowcomms_get_buffer(int nodeid, int len, gfp_t allocation, char **ppc)
  1137. {
  1138. struct connection *con;
  1139. struct writequeue_entry *e;
  1140. int offset = 0;
  1141. con = nodeid2con(nodeid, allocation);
  1142. if (!con)
  1143. return NULL;
  1144. spin_lock(&con->writequeue_lock);
  1145. e = list_entry(con->writequeue.prev, struct writequeue_entry, list);
  1146. if ((&e->list == &con->writequeue) ||
  1147. (PAGE_SIZE - e->end < len)) {
  1148. e = NULL;
  1149. } else {
  1150. offset = e->end;
  1151. e->end += len;
  1152. e->users++;
  1153. }
  1154. spin_unlock(&con->writequeue_lock);
  1155. if (e) {
  1156. got_one:
  1157. *ppc = page_address(e->page) + offset;
  1158. return e;
  1159. }
  1160. e = new_writequeue_entry(con, allocation);
  1161. if (e) {
  1162. spin_lock(&con->writequeue_lock);
  1163. offset = e->end;
  1164. e->end += len;
  1165. e->users++;
  1166. list_add_tail(&e->list, &con->writequeue);
  1167. spin_unlock(&con->writequeue_lock);
  1168. goto got_one;
  1169. }
  1170. return NULL;
  1171. }
  1172. void dlm_lowcomms_commit_buffer(void *mh)
  1173. {
  1174. struct writequeue_entry *e = (struct writequeue_entry *)mh;
  1175. struct connection *con = e->con;
  1176. int users;
  1177. spin_lock(&con->writequeue_lock);
  1178. users = --e->users;
  1179. if (users)
  1180. goto out;
  1181. e->len = e->end - e->offset;
  1182. spin_unlock(&con->writequeue_lock);
  1183. queue_work(send_workqueue, &con->swork);
  1184. return;
  1185. out:
  1186. spin_unlock(&con->writequeue_lock);
  1187. return;
  1188. }
  1189. /* Send a message */
  1190. static void send_to_sock(struct connection *con)
  1191. {
  1192. int ret = 0;
  1193. const int msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL;
  1194. struct writequeue_entry *e;
  1195. int len, offset;
  1196. int count = 0;
  1197. mutex_lock(&con->sock_mutex);
  1198. if (con->sock == NULL)
  1199. goto out_connect;
  1200. spin_lock(&con->writequeue_lock);
  1201. for (;;) {
  1202. e = list_entry(con->writequeue.next, struct writequeue_entry,
  1203. list);
  1204. if ((struct list_head *) e == &con->writequeue)
  1205. break;
  1206. len = e->len;
  1207. offset = e->offset;
  1208. BUG_ON(len == 0 && e->users == 0);
  1209. spin_unlock(&con->writequeue_lock);
  1210. ret = 0;
  1211. if (len) {
  1212. ret = kernel_sendpage(con->sock, e->page, offset, len,
  1213. msg_flags);
  1214. if (ret == -EAGAIN || ret == 0) {
  1215. if (ret == -EAGAIN &&
  1216. test_bit(SOCKWQ_ASYNC_NOSPACE, &con->sock->flags) &&
  1217. !test_and_set_bit(CF_APP_LIMITED, &con->flags)) {
  1218. /* Notify TCP that we're limited by the
  1219. * application window size.
  1220. */
  1221. set_bit(SOCK_NOSPACE, &con->sock->flags);
  1222. con->sock->sk->sk_write_pending++;
  1223. }
  1224. cond_resched();
  1225. goto out;
  1226. } else if (ret < 0)
  1227. goto send_error;
  1228. }
  1229. /* Don't starve people filling buffers */
  1230. if (++count >= MAX_SEND_MSG_COUNT) {
  1231. cond_resched();
  1232. count = 0;
  1233. }
  1234. spin_lock(&con->writequeue_lock);
  1235. writequeue_entry_complete(e, ret);
  1236. }
  1237. spin_unlock(&con->writequeue_lock);
  1238. out:
  1239. mutex_unlock(&con->sock_mutex);
  1240. return;
  1241. send_error:
  1242. mutex_unlock(&con->sock_mutex);
  1243. close_connection(con, false, false, true);
  1244. /* Requeue the send work. When the work daemon runs again, it will try
  1245. a new connection, then call this function again. */
  1246. queue_work(send_workqueue, &con->swork);
  1247. return;
  1248. out_connect:
  1249. mutex_unlock(&con->sock_mutex);
  1250. queue_work(send_workqueue, &con->swork);
  1251. cond_resched();
  1252. }
  1253. static void clean_one_writequeue(struct connection *con)
  1254. {
  1255. struct writequeue_entry *e, *safe;
  1256. spin_lock(&con->writequeue_lock);
  1257. list_for_each_entry_safe(e, safe, &con->writequeue, list) {
  1258. list_del(&e->list);
  1259. free_entry(e);
  1260. }
  1261. spin_unlock(&con->writequeue_lock);
  1262. }
  1263. /* Called from recovery when it knows that a node has
  1264. left the cluster */
  1265. int dlm_lowcomms_close(int nodeid)
  1266. {
  1267. struct connection *con;
  1268. struct dlm_node_addr *na;
  1269. log_print("closing connection to node %d", nodeid);
  1270. con = nodeid2con(nodeid, 0);
  1271. if (con) {
  1272. set_bit(CF_CLOSE, &con->flags);
  1273. close_connection(con, true, true, true);
  1274. clean_one_writequeue(con);
  1275. }
  1276. spin_lock(&dlm_node_addrs_spin);
  1277. na = find_node_addr(nodeid);
  1278. if (na) {
  1279. list_del(&na->list);
  1280. while (na->addr_count--)
  1281. kfree(na->addr[na->addr_count]);
  1282. kfree(na);
  1283. }
  1284. spin_unlock(&dlm_node_addrs_spin);
  1285. return 0;
  1286. }
  1287. /* Receive workqueue function */
  1288. static void process_recv_sockets(struct work_struct *work)
  1289. {
  1290. struct connection *con = container_of(work, struct connection, rwork);
  1291. int err;
  1292. clear_bit(CF_READ_PENDING, &con->flags);
  1293. do {
  1294. err = con->rx_action(con);
  1295. } while (!err);
  1296. }
  1297. /* Send workqueue function */
  1298. static void process_send_sockets(struct work_struct *work)
  1299. {
  1300. struct connection *con = container_of(work, struct connection, swork);
  1301. clear_bit(CF_WRITE_PENDING, &con->flags);
  1302. if (con->sock == NULL) /* not mutex protected so check it inside too */
  1303. con->connect_action(con);
  1304. if (!list_empty(&con->writequeue))
  1305. send_to_sock(con);
  1306. }
  1307. static void work_stop(void)
  1308. {
  1309. if (recv_workqueue)
  1310. destroy_workqueue(recv_workqueue);
  1311. if (send_workqueue)
  1312. destroy_workqueue(send_workqueue);
  1313. }
  1314. static int work_start(void)
  1315. {
  1316. recv_workqueue = alloc_workqueue("dlm_recv",
  1317. WQ_UNBOUND | WQ_MEM_RECLAIM, 1);
  1318. if (!recv_workqueue) {
  1319. log_print("can't start dlm_recv");
  1320. return -ENOMEM;
  1321. }
  1322. send_workqueue = alloc_workqueue("dlm_send",
  1323. WQ_UNBOUND | WQ_MEM_RECLAIM, 1);
  1324. if (!send_workqueue) {
  1325. log_print("can't start dlm_send");
  1326. destroy_workqueue(recv_workqueue);
  1327. return -ENOMEM;
  1328. }
  1329. return 0;
  1330. }
  1331. static void _stop_conn(struct connection *con, bool and_other)
  1332. {
  1333. mutex_lock(&con->sock_mutex);
  1334. set_bit(CF_CLOSE, &con->flags);
  1335. set_bit(CF_READ_PENDING, &con->flags);
  1336. set_bit(CF_WRITE_PENDING, &con->flags);
  1337. if (con->sock && con->sock->sk) {
  1338. write_lock_bh(&con->sock->sk->sk_callback_lock);
  1339. con->sock->sk->sk_user_data = NULL;
  1340. write_unlock_bh(&con->sock->sk->sk_callback_lock);
  1341. }
  1342. if (con->othercon && and_other)
  1343. _stop_conn(con->othercon, false);
  1344. mutex_unlock(&con->sock_mutex);
  1345. }
  1346. static void stop_conn(struct connection *con)
  1347. {
  1348. _stop_conn(con, true);
  1349. }
  1350. static void shutdown_conn(struct connection *con)
  1351. {
  1352. if (con->shutdown_action)
  1353. con->shutdown_action(con);
  1354. }
  1355. static void connection_release(struct rcu_head *rcu)
  1356. {
  1357. struct connection *con = container_of(rcu, struct connection, rcu);
  1358. kfree(con->rx_buf);
  1359. kfree(con);
  1360. }
  1361. static void free_conn(struct connection *con)
  1362. {
  1363. close_connection(con, true, true, true);
  1364. spin_lock(&connections_lock);
  1365. hlist_del_rcu(&con->list);
  1366. spin_unlock(&connections_lock);
  1367. if (con->othercon) {
  1368. clean_one_writequeue(con->othercon);
  1369. call_rcu(&con->othercon->rcu, connection_release);
  1370. }
  1371. clean_one_writequeue(con);
  1372. call_rcu(&con->rcu, connection_release);
  1373. }
  1374. static void work_flush(void)
  1375. {
  1376. int ok, idx;
  1377. int i;
  1378. struct connection *con;
  1379. do {
  1380. ok = 1;
  1381. foreach_conn(stop_conn);
  1382. if (recv_workqueue)
  1383. flush_workqueue(recv_workqueue);
  1384. if (send_workqueue)
  1385. flush_workqueue(send_workqueue);
  1386. idx = srcu_read_lock(&connections_srcu);
  1387. for (i = 0; i < CONN_HASH_SIZE && ok; i++) {
  1388. hlist_for_each_entry_rcu(con, &connection_hash[i],
  1389. list) {
  1390. ok &= test_bit(CF_READ_PENDING, &con->flags);
  1391. ok &= test_bit(CF_WRITE_PENDING, &con->flags);
  1392. if (con->othercon) {
  1393. ok &= test_bit(CF_READ_PENDING,
  1394. &con->othercon->flags);
  1395. ok &= test_bit(CF_WRITE_PENDING,
  1396. &con->othercon->flags);
  1397. }
  1398. }
  1399. }
  1400. srcu_read_unlock(&connections_srcu, idx);
  1401. } while (!ok);
  1402. }
  1403. void dlm_lowcomms_stop(void)
  1404. {
  1405. /* Set all the flags to prevent any
  1406. socket activity.
  1407. */
  1408. dlm_allow_conn = 0;
  1409. if (recv_workqueue)
  1410. flush_workqueue(recv_workqueue);
  1411. if (send_workqueue)
  1412. flush_workqueue(send_workqueue);
  1413. foreach_conn(shutdown_conn);
  1414. work_flush();
  1415. foreach_conn(free_conn);
  1416. work_stop();
  1417. deinit_local();
  1418. }
  1419. int dlm_lowcomms_start(void)
  1420. {
  1421. int error = -EINVAL;
  1422. struct connection *con;
  1423. int i;
  1424. for (i = 0; i < CONN_HASH_SIZE; i++)
  1425. INIT_HLIST_HEAD(&connection_hash[i]);
  1426. init_local();
  1427. if (!dlm_local_count) {
  1428. error = -ENOTCONN;
  1429. log_print("no local IP address has been set");
  1430. goto fail;
  1431. }
  1432. error = work_start();
  1433. if (error)
  1434. goto fail;
  1435. dlm_allow_conn = 1;
  1436. /* Start listening */
  1437. if (dlm_config.ci_protocol == 0)
  1438. error = tcp_listen_for_all();
  1439. else
  1440. error = sctp_listen_for_all();
  1441. if (error)
  1442. goto fail_unlisten;
  1443. return 0;
  1444. fail_unlisten:
  1445. dlm_allow_conn = 0;
  1446. con = nodeid2con(0,0);
  1447. if (con)
  1448. free_conn(con);
  1449. fail:
  1450. return error;
  1451. }
  1452. void dlm_lowcomms_exit(void)
  1453. {
  1454. struct dlm_node_addr *na, *safe;
  1455. spin_lock(&dlm_node_addrs_spin);
  1456. list_for_each_entry_safe(na, safe, &dlm_node_addrs, list) {
  1457. list_del(&na->list);
  1458. while (na->addr_count--)
  1459. kfree(na->addr[na->addr_count]);
  1460. kfree(na);
  1461. }
  1462. spin_unlock(&dlm_node_addrs_spin);
  1463. }