threads.c 8.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311
  1. /*
  2. * Copyright (c) 2006, 2018 Oracle and/or its affiliates. All rights reserved.
  3. *
  4. * This software is available to you under a choice of one of two
  5. * licenses. You may choose to be licensed under the terms of the GNU
  6. * General Public License (GPL) Version 2, available from the file
  7. * COPYING in the main directory of this source tree, or the
  8. * OpenIB.org BSD license below:
  9. *
  10. * Redistribution and use in source and binary forms, with or
  11. * without modification, are permitted provided that the following
  12. * conditions are met:
  13. *
  14. * - Redistributions of source code must retain the above
  15. * copyright notice, this list of conditions and the following
  16. * disclaimer.
  17. *
  18. * - Redistributions in binary form must reproduce the above
  19. * copyright notice, this list of conditions and the following
  20. * disclaimer in the documentation and/or other materials
  21. * provided with the distribution.
  22. *
  23. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
  24. * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  25. * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  26. * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
  27. * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
  28. * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
  29. * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  30. * SOFTWARE.
  31. *
  32. */
  33. #include <linux/kernel.h>
  34. #include <linux/random.h>
  35. #include <linux/export.h>
  36. #include "rds.h"
  37. /*
  38. * All of connection management is simplified by serializing it through
  39. * work queues that execute in a connection managing thread.
  40. *
  41. * TCP wants to send acks through sendpage() in response to data_ready(),
  42. * but it needs a process context to do so.
  43. *
  44. * The receive paths need to allocate but can't drop packets (!) so we have
  45. * a thread around to block allocating if the receive fast path sees an
  46. * allocation failure.
  47. */
  48. /* Grand Unified Theory of connection life cycle:
  49. * At any point in time, the connection can be in one of these states:
  50. * DOWN, CONNECTING, UP, DISCONNECTING, ERROR
  51. *
  52. * The following transitions are possible:
  53. * ANY -> ERROR
  54. * UP -> DISCONNECTING
  55. * ERROR -> DISCONNECTING
  56. * DISCONNECTING -> DOWN
  57. * DOWN -> CONNECTING
  58. * CONNECTING -> UP
  59. *
  60. * Transition to state DISCONNECTING/DOWN:
  61. * - Inside the shutdown worker; synchronizes with xmit path
  62. * through RDS_IN_XMIT, and with connection management callbacks
  63. * via c_cm_lock.
  64. *
  65. * For receive callbacks, we rely on the underlying transport
  66. * (TCP, IB/RDMA) to provide the necessary synchronisation.
  67. */
  68. struct workqueue_struct *rds_wq;
  69. EXPORT_SYMBOL_GPL(rds_wq);
  70. void rds_connect_path_complete(struct rds_conn_path *cp, int curr)
  71. {
  72. if (!rds_conn_path_transition(cp, curr, RDS_CONN_UP)) {
  73. printk(KERN_WARNING "%s: Cannot transition to state UP, "
  74. "current state is %d\n",
  75. __func__,
  76. atomic_read(&cp->cp_state));
  77. rds_conn_path_drop(cp, false);
  78. return;
  79. }
  80. rdsdebug("conn %p for %pI6c to %pI6c complete\n",
  81. cp->cp_conn, &cp->cp_conn->c_laddr, &cp->cp_conn->c_faddr);
  82. cp->cp_reconnect_jiffies = 0;
  83. set_bit(0, &cp->cp_conn->c_map_queued);
  84. rcu_read_lock();
  85. if (!rds_destroy_pending(cp->cp_conn)) {
  86. queue_delayed_work(rds_wq, &cp->cp_send_w, 0);
  87. queue_delayed_work(rds_wq, &cp->cp_recv_w, 0);
  88. }
  89. rcu_read_unlock();
  90. cp->cp_conn->c_proposed_version = RDS_PROTOCOL_VERSION;
  91. }
  92. EXPORT_SYMBOL_GPL(rds_connect_path_complete);
  93. void rds_connect_complete(struct rds_connection *conn)
  94. {
  95. rds_connect_path_complete(&conn->c_path[0], RDS_CONN_CONNECTING);
  96. }
  97. EXPORT_SYMBOL_GPL(rds_connect_complete);
  98. /*
  99. * This random exponential backoff is relied on to eventually resolve racing
  100. * connects.
  101. *
  102. * If connect attempts race then both parties drop both connections and come
  103. * here to wait for a random amount of time before trying again. Eventually
  104. * the backoff range will be so much greater than the time it takes to
  105. * establish a connection that one of the pair will establish the connection
  106. * before the other's random delay fires.
  107. *
  108. * Connection attempts that arrive while a connection is already established
  109. * are also considered to be racing connects. This lets a connection from
  110. * a rebooted machine replace an existing stale connection before the transport
  111. * notices that the connection has failed.
  112. *
  113. * We should *always* start with a random backoff; otherwise a broken connection
  114. * will always take several iterations to be re-established.
  115. */
  116. void rds_queue_reconnect(struct rds_conn_path *cp)
  117. {
  118. unsigned long rand;
  119. struct rds_connection *conn = cp->cp_conn;
  120. rdsdebug("conn %p for %pI6c to %pI6c reconnect jiffies %lu\n",
  121. conn, &conn->c_laddr, &conn->c_faddr,
  122. cp->cp_reconnect_jiffies);
  123. /* let peer with smaller addr initiate reconnect, to avoid duels */
  124. if (conn->c_trans->t_type == RDS_TRANS_TCP &&
  125. rds_addr_cmp(&conn->c_laddr, &conn->c_faddr) >= 0)
  126. return;
  127. set_bit(RDS_RECONNECT_PENDING, &cp->cp_flags);
  128. if (cp->cp_reconnect_jiffies == 0) {
  129. cp->cp_reconnect_jiffies = rds_sysctl_reconnect_min_jiffies;
  130. rcu_read_lock();
  131. if (!rds_destroy_pending(cp->cp_conn))
  132. queue_delayed_work(rds_wq, &cp->cp_conn_w, 0);
  133. rcu_read_unlock();
  134. return;
  135. }
  136. get_random_bytes(&rand, sizeof(rand));
  137. rdsdebug("%lu delay %lu ceil conn %p for %pI6c -> %pI6c\n",
  138. rand % cp->cp_reconnect_jiffies, cp->cp_reconnect_jiffies,
  139. conn, &conn->c_laddr, &conn->c_faddr);
  140. rcu_read_lock();
  141. if (!rds_destroy_pending(cp->cp_conn))
  142. queue_delayed_work(rds_wq, &cp->cp_conn_w,
  143. rand % cp->cp_reconnect_jiffies);
  144. rcu_read_unlock();
  145. cp->cp_reconnect_jiffies = min(cp->cp_reconnect_jiffies * 2,
  146. rds_sysctl_reconnect_max_jiffies);
  147. }
  148. void rds_connect_worker(struct work_struct *work)
  149. {
  150. struct rds_conn_path *cp = container_of(work,
  151. struct rds_conn_path,
  152. cp_conn_w.work);
  153. struct rds_connection *conn = cp->cp_conn;
  154. int ret;
  155. if (cp->cp_index > 0 &&
  156. rds_addr_cmp(&cp->cp_conn->c_laddr, &cp->cp_conn->c_faddr) >= 0)
  157. return;
  158. clear_bit(RDS_RECONNECT_PENDING, &cp->cp_flags);
  159. ret = rds_conn_path_transition(cp, RDS_CONN_DOWN, RDS_CONN_CONNECTING);
  160. if (ret) {
  161. ret = conn->c_trans->conn_path_connect(cp);
  162. rdsdebug("conn %p for %pI6c to %pI6c dispatched, ret %d\n",
  163. conn, &conn->c_laddr, &conn->c_faddr, ret);
  164. if (ret) {
  165. if (rds_conn_path_transition(cp,
  166. RDS_CONN_CONNECTING,
  167. RDS_CONN_DOWN))
  168. rds_queue_reconnect(cp);
  169. else
  170. rds_conn_path_error(cp, "connect failed\n");
  171. }
  172. }
  173. }
  174. void rds_send_worker(struct work_struct *work)
  175. {
  176. struct rds_conn_path *cp = container_of(work,
  177. struct rds_conn_path,
  178. cp_send_w.work);
  179. int ret;
  180. if (rds_conn_path_state(cp) == RDS_CONN_UP) {
  181. clear_bit(RDS_LL_SEND_FULL, &cp->cp_flags);
  182. ret = rds_send_xmit(cp);
  183. cond_resched();
  184. rdsdebug("conn %p ret %d\n", cp->cp_conn, ret);
  185. switch (ret) {
  186. case -EAGAIN:
  187. rds_stats_inc(s_send_immediate_retry);
  188. queue_delayed_work(rds_wq, &cp->cp_send_w, 0);
  189. break;
  190. case -ENOMEM:
  191. rds_stats_inc(s_send_delayed_retry);
  192. queue_delayed_work(rds_wq, &cp->cp_send_w, 2);
  193. break;
  194. default:
  195. break;
  196. }
  197. }
  198. }
  199. void rds_recv_worker(struct work_struct *work)
  200. {
  201. struct rds_conn_path *cp = container_of(work,
  202. struct rds_conn_path,
  203. cp_recv_w.work);
  204. int ret;
  205. if (rds_conn_path_state(cp) == RDS_CONN_UP) {
  206. ret = cp->cp_conn->c_trans->recv_path(cp);
  207. rdsdebug("conn %p ret %d\n", cp->cp_conn, ret);
  208. switch (ret) {
  209. case -EAGAIN:
  210. rds_stats_inc(s_recv_immediate_retry);
  211. queue_delayed_work(rds_wq, &cp->cp_recv_w, 0);
  212. break;
  213. case -ENOMEM:
  214. rds_stats_inc(s_recv_delayed_retry);
  215. queue_delayed_work(rds_wq, &cp->cp_recv_w, 2);
  216. break;
  217. default:
  218. break;
  219. }
  220. }
  221. }
  222. void rds_shutdown_worker(struct work_struct *work)
  223. {
  224. struct rds_conn_path *cp = container_of(work,
  225. struct rds_conn_path,
  226. cp_down_w);
  227. rds_conn_shutdown(cp);
  228. }
  229. void rds_threads_exit(void)
  230. {
  231. destroy_workqueue(rds_wq);
  232. }
  233. int rds_threads_init(void)
  234. {
  235. rds_wq = create_singlethread_workqueue("krdsd");
  236. if (!rds_wq)
  237. return -ENOMEM;
  238. return 0;
  239. }
  240. /* Compare two IPv6 addresses. Return 0 if the two addresses are equal.
  241. * Return 1 if the first is greater. Return -1 if the second is greater.
  242. */
  243. int rds_addr_cmp(const struct in6_addr *addr1,
  244. const struct in6_addr *addr2)
  245. {
  246. #if defined(CONFIG_HAVE_EFFICIENT_UNALIGNED_ACCESS) && BITS_PER_LONG == 64
  247. const __be64 *a1, *a2;
  248. u64 x, y;
  249. a1 = (__be64 *)addr1;
  250. a2 = (__be64 *)addr2;
  251. if (*a1 != *a2) {
  252. if (be64_to_cpu(*a1) < be64_to_cpu(*a2))
  253. return -1;
  254. else
  255. return 1;
  256. } else {
  257. x = be64_to_cpu(*++a1);
  258. y = be64_to_cpu(*++a2);
  259. if (x < y)
  260. return -1;
  261. else if (x > y)
  262. return 1;
  263. else
  264. return 0;
  265. }
  266. #else
  267. u32 a, b;
  268. int i;
  269. for (i = 0; i < 4; i++) {
  270. if (addr1->s6_addr32[i] != addr2->s6_addr32[i]) {
  271. a = ntohl(addr1->s6_addr32[i]);
  272. b = ntohl(addr2->s6_addr32[i]);
  273. if (a < b)
  274. return -1;
  275. else if (a > b)
  276. return 1;
  277. }
  278. }
  279. return 0;
  280. #endif
  281. }
  282. EXPORT_SYMBOL_GPL(rds_addr_cmp);