lowcomms.c 42 KB

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