140 	struct connection *con;  member
181 	struct connection *con;  in __find_con()  local
185 	hlist_for_each_entry(con, &connection_hash[r], list) {  in __find_con()
186 		if (con->nodeid == nodeid)  in __find_con()
187 			return con;  in __find_con()
198 	struct connection *con = NULL;  in __nodeid2con()  local
201 	con = __find_con(nodeid);  in __nodeid2con()
202 	if (con || !alloc)  in __nodeid2con()
203 		return con;  in __nodeid2con()
205 	con = kmem_cache_zalloc(con_cache, alloc);  in __nodeid2con()
206 	if (!con)  in __nodeid2con()
210 	hlist_add_head(&con->list, &connection_hash[r]);  in __nodeid2con()
212 	con->nodeid = nodeid;  in __nodeid2con()
213 	mutex_init(&con->sock_mutex);  in __nodeid2con()
214 	INIT_LIST_HEAD(&con->writequeue);  in __nodeid2con()
215 	spin_lock_init(&con->writequeue_lock);  in __nodeid2con()
216 	INIT_WORK(&con->swork, process_send_sockets);  in __nodeid2con()
217 	INIT_WORK(&con->rwork, process_recv_sockets);  in __nodeid2con()
220 	if (con->nodeid) {  in __nodeid2con()
223 		con->connect_action = zerocon->connect_action;  in __nodeid2con()
224 		if (!con->rx_action)  in __nodeid2con()
225 			con->rx_action = zerocon->rx_action;  in __nodeid2con()
228 	return con;  in __nodeid2con()
236 	struct connection *con;  in foreach_conn()  local
239 		hlist_for_each_entry_safe(con, n, &connection_hash[i], list)  in foreach_conn()
240 			conn_func(con);  in foreach_conn()
246 	struct connection *con;  in nodeid2con()  local
249 	con = __nodeid2con(nodeid, allocation);  in nodeid2con()
252 	return con;  in nodeid2con()
259 	struct connection *con;  in assoc2con()  local
264 		hlist_for_each_entry(con, &connection_hash[i], list) {  in assoc2con()
265 			if (con->sctp_assoc == assoc_id) {  in assoc2con()
267 				return con;  in assoc2con()
429 	struct connection *con = sock2con(sk);  in lowcomms_data_ready()  local
430 	if (con && !test_and_set_bit(CF_READ_PENDING, &con->flags))  in lowcomms_data_ready()
431 		queue_work(recv_workqueue, &con->rwork);  in lowcomms_data_ready()
436 	struct connection *con = sock2con(sk);  in lowcomms_write_space()  local
438 	if (!con)  in lowcomms_write_space()
441 	clear_bit(SOCK_NOSPACE, &con->sock->flags);  in lowcomms_write_space()
443 	if (test_and_clear_bit(CF_APP_LIMITED, &con->flags)) {  in lowcomms_write_space()
444 		con->sock->sk->sk_write_pending--;  in lowcomms_write_space()
445 		clear_bit(SOCK_ASYNC_NOSPACE, &con->sock->flags);  in lowcomms_write_space()
448 	if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags))  in lowcomms_write_space()
449 		queue_work(send_workqueue, &con->swork);  in lowcomms_write_space()
452 static inline void lowcomms_connect_sock(struct connection *con)  in lowcomms_connect_sock()  argument
454 	if (test_bit(CF_CLOSE, &con->flags))  in lowcomms_connect_sock()
456 	if (!test_and_set_bit(CF_CONNECT_PENDING, &con->flags))  in lowcomms_connect_sock()
457 		queue_work(send_workqueue, &con->swork);  in lowcomms_connect_sock()
468 	struct connection *con;  in dlm_lowcomms_connect_node()  local
477 	con = nodeid2con(nodeid, GFP_NOFS);  in dlm_lowcomms_connect_node()
478 	if (!con)  in dlm_lowcomms_connect_node()
480 	lowcomms_connect_sock(con);  in dlm_lowcomms_connect_node()
485 static void add_sock(struct socket *sock, struct connection *con)  in add_sock()  argument
487 	con->sock = sock;  in add_sock()
490 	con->sock->sk->sk_data_ready = lowcomms_data_ready;  in add_sock()
491 	con->sock->sk->sk_write_space = lowcomms_write_space;  in add_sock()
492 	con->sock->sk->sk_state_change = lowcomms_state_change;  in add_sock()
493 	con->sock->sk->sk_user_data = con;  in add_sock()
494 	con->sock->sk->sk_allocation = GFP_NOFS;  in add_sock()
517 static void close_connection(struct connection *con, bool and_other)  in close_connection()  argument
519 	mutex_lock(&con->sock_mutex);  in close_connection()
521 	if (con->sock) {  in close_connection()
522 		sock_release(con->sock);  in close_connection()
523 		con->sock = NULL;  in close_connection()
525 	if (con->othercon && and_other) {  in close_connection()
527 		close_connection(con->othercon, false);  in close_connection()
529 	if (con->rx_page) {  in close_connection()
530 		__free_page(con->rx_page);  in close_connection()
531 		con->rx_page = NULL;  in close_connection()
534 	con->retries = 0;  in close_connection()
535 	mutex_unlock(&con->sock_mutex);  in close_connection()
546 	struct connection *con;  in sctp_send_shutdown()  local
548 	con = nodeid2con(0,0);  in sctp_send_shutdown()
549 	BUG_ON(con == NULL);  in sctp_send_shutdown()
568 	ret = kernel_sendmsg(con->sock, &outmessage, NULL, 0, 0);  in sctp_send_shutdown()
574 static void sctp_init_failed_foreach(struct connection *con)  in sctp_init_failed_foreach()  argument
585 	if (!con->nodeid || con->sctp_assoc)  in sctp_init_failed_foreach()
588 	log_print("Retrying SCTP association init for node %d\n", con->nodeid);  in sctp_init_failed_foreach()
590 	con->try_new_addr = true;  in sctp_init_failed_foreach()
591 	con->sctp_assoc = 0;  in sctp_init_failed_foreach()
592 	if (test_and_clear_bit(CF_INIT_PENDING, &con->flags)) {  in sctp_init_failed_foreach()
593 		if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags))  in sctp_init_failed_foreach()
594 			queue_work(send_workqueue, &con->swork);  in sctp_init_failed_foreach()
615 	struct connection *con;  in retry_failed_sctp_send()  local
626 	con = nodeid2con(nodeid, 0);  in retry_failed_sctp_send()
627 	if (!con) {  in retry_failed_sctp_send()
645 	if (con->sctp_assoc) {  in retry_failed_sctp_send()
646 		if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags))  in retry_failed_sctp_send()
647 			queue_work(send_workqueue, &con->swork);  in retry_failed_sctp_send()
649 		sctp_init_failed_foreach(con);  in retry_failed_sctp_send()
653 static void process_sctp_notification(struct connection *con,  in process_sctp_notification()  argument
661 		retry_failed_sctp_send(con, &sn->sn_send_failed, buf);  in process_sctp_notification()
690 			ret = kernel_getsockopt(con->sock,  in process_sctp_notification()
704 					clear_bit(CF_CONNECT_PENDING, &con->flags);  in process_sctp_notification()
722 			lock_sock(con->sock->sk);  in process_sctp_notification()
723 			ret = sctp_do_peeloff(con->sock->sk,  in process_sctp_notification()
726 			release_sock(con->sock->sk);  in process_sctp_notification()
762 			con = assoc2con(sn->sn_assoc_change.sac_assoc_id);  in process_sctp_notification()
763 			if (con) {  in process_sctp_notification()
764 				con->sctp_assoc = 0;  in process_sctp_notification()
787 static int receive_from_sock(struct connection *con)  in receive_from_sock()  argument
798 	mutex_lock(&con->sock_mutex);  in receive_from_sock()
800 	if (con->sock == NULL) {  in receive_from_sock()
805 	if (con->rx_page == NULL) {  in receive_from_sock()
810 		con->rx_page = alloc_page(GFP_ATOMIC);  in receive_from_sock()
811 		if (con->rx_page == NULL)  in receive_from_sock()
813 		cbuf_init(&con->cb, PAGE_CACHE_SIZE);  in receive_from_sock()
825 	iov[0].iov_len = con->cb.base - cbuf_data(&con->cb);  in receive_from_sock()
826 	iov[0].iov_base = page_address(con->rx_page) + cbuf_data(&con->cb);  in receive_from_sock()
834 	if (cbuf_data(&con->cb) >= con->cb.base) {  in receive_from_sock()
835 		iov[0].iov_len = PAGE_CACHE_SIZE - cbuf_data(&con->cb);  in receive_from_sock()
836 		iov[1].iov_len = con->cb.base;  in receive_from_sock()
837 		iov[1].iov_base = page_address(con->rx_page);  in receive_from_sock()
842 	r = ret = kernel_recvmsg(con->sock, &msg, iov, nvec, len,  in receive_from_sock()
852 		process_sctp_notification(con, &msg,  in receive_from_sock()
853 				page_address(con->rx_page) + con->cb.base);  in receive_from_sock()
854 		mutex_unlock(&con->sock_mutex);  in receive_from_sock()
857 	BUG_ON(con->nodeid == 0);  in receive_from_sock()
861 	cbuf_add(&con->cb, ret);  in receive_from_sock()
862 	ret = dlm_process_incoming_buffer(con->nodeid,  in receive_from_sock()
863 					  page_address(con->rx_page),  in receive_from_sock()
864 					  con->cb.base, con->cb.len,  in receive_from_sock()
869 			  page_address(con->rx_page), con->cb.base, con->cb.len,  in receive_from_sock()
874 	cbuf_eat(&con->cb, ret);  in receive_from_sock()
876 	if (cbuf_empty(&con->cb) && !call_again_soon) {  in receive_from_sock()
877 		__free_page(con->rx_page);  in receive_from_sock()
878 		con->rx_page = NULL;  in receive_from_sock()
883 	mutex_unlock(&con->sock_mutex);  in receive_from_sock()
887 	if (!test_and_set_bit(CF_READ_PENDING, &con->flags))  in receive_from_sock()
888 		queue_work(recv_workqueue, &con->rwork);  in receive_from_sock()
889 	mutex_unlock(&con->sock_mutex);  in receive_from_sock()
893 	mutex_unlock(&con->sock_mutex);  in receive_from_sock()
895 		close_connection(con, false);  in receive_from_sock()
906 static int tcp_accept_from_sock(struct connection *con)  in tcp_accept_from_sock()  argument
929 	mutex_lock_nested(&con->sock_mutex, 0);  in tcp_accept_from_sock()
932 	if (con->sock == NULL)  in tcp_accept_from_sock()
935 	newsock->type = con->sock->type;  in tcp_accept_from_sock()
936 	newsock->ops = con->sock->ops;  in tcp_accept_from_sock()
938 	result = con->sock->ops->accept(con->sock, newsock, O_NONBLOCK);  in tcp_accept_from_sock()
958 		mutex_unlock(&con->sock_mutex);  in tcp_accept_from_sock()
1023 	mutex_unlock(&con->sock_mutex);  in tcp_accept_from_sock()
1028 	mutex_unlock(&con->sock_mutex);  in tcp_accept_from_sock()
1065 static void sctp_init_assoc(struct connection *con)  in sctp_init_assoc()  argument
1079 	mutex_lock(&con->sock_mutex);  in sctp_init_assoc()
1080 	if (test_and_set_bit(CF_INIT_PENDING, &con->flags))  in sctp_init_assoc()
1083 	if (nodeid_to_addr(con->nodeid, NULL, (struct sockaddr *)&rem_addr,  in sctp_init_assoc()
1084 			   con->try_new_addr)) {  in sctp_init_assoc()
1085 		log_print("no address for nodeid %d", con->nodeid);  in sctp_init_assoc()
1099 	spin_lock(&con->writequeue_lock);  in sctp_init_assoc()
1101 	if (list_empty(&con->writequeue)) {  in sctp_init_assoc()
1102 		spin_unlock(&con->writequeue_lock);  in sctp_init_assoc()
1103 		log_print("writequeue empty for nodeid %d", con->nodeid);  in sctp_init_assoc()
1107 	e = list_first_entry(&con->writequeue, struct writequeue_entry, list);  in sctp_init_assoc()
1114 	spin_unlock(&con->writequeue_lock);  in sctp_init_assoc()
1130 	sinfo->sinfo_ppid = cpu_to_le32(con->nodeid);  in sctp_init_assoc()
1137 			  con->nodeid, ret);  in sctp_init_assoc()
1140 		clear_bit(CF_CONNECT_PENDING, &con->flags);  in sctp_init_assoc()
1141 		clear_bit(CF_INIT_PENDING, &con->flags);  in sctp_init_assoc()
1144 		spin_lock(&con->writequeue_lock);  in sctp_init_assoc()
1146 		spin_unlock(&con->writequeue_lock);  in sctp_init_assoc()
1150 	mutex_unlock(&con->sock_mutex);  in sctp_init_assoc()
1154 static void tcp_connect_to_sock(struct connection *con)  in tcp_connect_to_sock()  argument
1162 	if (con->nodeid == 0) {  in tcp_connect_to_sock()
1167 	mutex_lock(&con->sock_mutex);  in tcp_connect_to_sock()
1168 	if (con->retries++ > MAX_CONNECT_RETRIES)  in tcp_connect_to_sock()
1172 	if (con->sock)  in tcp_connect_to_sock()
1182 	result = nodeid_to_addr(con->nodeid, &saddr, NULL, false);  in tcp_connect_to_sock()
1184 		log_print("no address for nodeid %d", con->nodeid);  in tcp_connect_to_sock()
1188 	sock->sk->sk_user_data = con;  in tcp_connect_to_sock()
1189 	con->rx_action = receive_from_sock;  in tcp_connect_to_sock()
1190 	con->connect_action = tcp_connect_to_sock;  in tcp_connect_to_sock()
1191 	add_sock(sock, con);  in tcp_connect_to_sock()
1206 	log_print("connecting to %d", con->nodeid);  in tcp_connect_to_sock()
1220 	if (con->sock) {  in tcp_connect_to_sock()
1221 		sock_release(con->sock);  in tcp_connect_to_sock()
1222 		con->sock = NULL;  in tcp_connect_to_sock()
1235 		log_print("connect %d try %d error %d", con->nodeid,  in tcp_connect_to_sock()
1236 			  con->retries, result);  in tcp_connect_to_sock()
1237 		mutex_unlock(&con->sock_mutex);  in tcp_connect_to_sock()
1239 		lowcomms_connect_sock(con);  in tcp_connect_to_sock()
1243 	mutex_unlock(&con->sock_mutex);  in tcp_connect_to_sock()
1247 static struct socket *tcp_create_listen_sock(struct connection *con,  in tcp_create_listen_sock()  argument
1278 	con->rx_action = tcp_accept_from_sock;  in tcp_create_listen_sock()
1279 	con->connect_action = tcp_connect_to_sock;  in tcp_create_listen_sock()
1288 		con->sock = NULL;  in tcp_create_listen_sock()
1359 	struct connection *con = nodeid2con(0, GFP_NOFS);  in sctp_listen_for_all()  local
1363 	if (!con)  in sctp_listen_for_all()
1402 	sock->sk->sk_user_data = con;  in sctp_listen_for_all()
1403 	con->sock = sock;  in sctp_listen_for_all()
1404 	con->sock->sk->sk_data_ready = lowcomms_data_ready;  in sctp_listen_for_all()
1405 	con->rx_action = receive_from_sock;  in sctp_listen_for_all()
1406 	con->connect_action = sctp_init_assoc;  in sctp_listen_for_all()
1413 		result = add_sctp_bind_addr(con, &localaddr, addr_len, num);  in sctp_listen_for_all()
1429 	con->sock = NULL;  in sctp_listen_for_all()
1437 	struct connection *con = nodeid2con(0, GFP_NOFS);  in tcp_listen_for_all()  local
1440 	if (!con)  in tcp_listen_for_all()
1452 	sock = tcp_create_listen_sock(con, dlm_local_addr[0]);  in tcp_listen_for_all()
1454 		add_sock(sock, con);  in tcp_listen_for_all()
1466 static struct writequeue_entry *new_writequeue_entry(struct connection *con,  in new_writequeue_entry()  argument
1485 	entry->con = con;  in new_writequeue_entry()
1492 	struct connection *con;  in dlm_lowcomms_get_buffer()  local
1496 	con = nodeid2con(nodeid, allocation);  in dlm_lowcomms_get_buffer()
1497 	if (!con)  in dlm_lowcomms_get_buffer()
1500 	spin_lock(&con->writequeue_lock);  in dlm_lowcomms_get_buffer()
1501 	e = list_entry(con->writequeue.prev, struct writequeue_entry, list);  in dlm_lowcomms_get_buffer()
1502 	if ((&e->list == &con->writequeue) ||  in dlm_lowcomms_get_buffer()
1510 	spin_unlock(&con->writequeue_lock);  in dlm_lowcomms_get_buffer()
1518 	e = new_writequeue_entry(con, allocation);  in dlm_lowcomms_get_buffer()
1520 		spin_lock(&con->writequeue_lock);  in dlm_lowcomms_get_buffer()
1524 		list_add_tail(&e->list, &con->writequeue);  in dlm_lowcomms_get_buffer()
1525 		spin_unlock(&con->writequeue_lock);  in dlm_lowcomms_get_buffer()
1534 	struct connection *con = e->con;  in dlm_lowcomms_commit_buffer()  local
1537 	spin_lock(&con->writequeue_lock);  in dlm_lowcomms_commit_buffer()
1542 	spin_unlock(&con->writequeue_lock);  in dlm_lowcomms_commit_buffer()
1544 	if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags)) {  in dlm_lowcomms_commit_buffer()
1545 		queue_work(send_workqueue, &con->swork);  in dlm_lowcomms_commit_buffer()
1550 	spin_unlock(&con->writequeue_lock);  in dlm_lowcomms_commit_buffer()
1555 static void send_to_sock(struct connection *con)  in send_to_sock()  argument
1563 	mutex_lock(&con->sock_mutex);  in send_to_sock()
1564 	if (con->sock == NULL)  in send_to_sock()
1567 	spin_lock(&con->writequeue_lock);  in send_to_sock()
1569 		e = list_entry(con->writequeue.next, struct writequeue_entry,  in send_to_sock()
1571 		if ((struct list_head *) e == &con->writequeue)  in send_to_sock()
1577 		spin_unlock(&con->writequeue_lock);  in send_to_sock()
1581 			ret = kernel_sendpage(con->sock, e->page, offset, len,  in send_to_sock()
1585 				    test_bit(SOCK_ASYNC_NOSPACE, &con->sock->flags) &&  in send_to_sock()
1586 				    !test_and_set_bit(CF_APP_LIMITED, &con->flags)) {  in send_to_sock()
1590 					set_bit(SOCK_NOSPACE, &con->sock->flags);  in send_to_sock()
1591 					con->sock->sk->sk_write_pending++;  in send_to_sock()
1605 		spin_lock(&con->writequeue_lock);  in send_to_sock()
1608 	spin_unlock(&con->writequeue_lock);  in send_to_sock()
1610 	mutex_unlock(&con->sock_mutex);  in send_to_sock()
1614 	mutex_unlock(&con->sock_mutex);  in send_to_sock()
1615 	close_connection(con, false);  in send_to_sock()
1616 	lowcomms_connect_sock(con);  in send_to_sock()
1620 	mutex_unlock(&con->sock_mutex);  in send_to_sock()
1621 	if (!test_bit(CF_INIT_PENDING, &con->flags))  in send_to_sock()
1622 		lowcomms_connect_sock(con);  in send_to_sock()
1625 static void clean_one_writequeue(struct connection *con)  in clean_one_writequeue()  argument
1629 	spin_lock(&con->writequeue_lock);  in clean_one_writequeue()
1630 	list_for_each_entry_safe(e, safe, &con->writequeue, list) {  in clean_one_writequeue()
1634 	spin_unlock(&con->writequeue_lock);  in clean_one_writequeue()
1641 	struct connection *con;  in dlm_lowcomms_close()  local
1645 	con = nodeid2con(nodeid, 0);  in dlm_lowcomms_close()
1646 	if (con) {  in dlm_lowcomms_close()
1647 		clear_bit(CF_CONNECT_PENDING, &con->flags);  in dlm_lowcomms_close()
1648 		clear_bit(CF_WRITE_PENDING, &con->flags);  in dlm_lowcomms_close()
1649 		set_bit(CF_CLOSE, &con->flags);  in dlm_lowcomms_close()
1650 		if (cancel_work_sync(&con->swork))  in dlm_lowcomms_close()
1652 		if (cancel_work_sync(&con->rwork))  in dlm_lowcomms_close()
1654 		clean_one_writequeue(con);  in dlm_lowcomms_close()
1655 		close_connection(con, true);  in dlm_lowcomms_close()
1674 	struct connection *con = container_of(work, struct connection, rwork);  in process_recv_sockets()  local
1677 	clear_bit(CF_READ_PENDING, &con->flags);  in process_recv_sockets()
1679 		err = con->rx_action(con);  in process_recv_sockets()
1686 	struct connection *con = container_of(work, struct connection, swork);  in process_send_sockets()  local
1688 	if (test_and_clear_bit(CF_CONNECT_PENDING, &con->flags)) {  in process_send_sockets()
1689 		con->connect_action(con);  in process_send_sockets()
1690 		set_bit(CF_WRITE_PENDING, &con->flags);  in process_send_sockets()
1692 	if (test_and_clear_bit(CF_WRITE_PENDING, &con->flags))  in process_send_sockets()
1693 		send_to_sock(con);  in process_send_sockets()
1729 static void stop_conn(struct connection *con)  in stop_conn()  argument
1731 	con->flags |= 0x0F;  in stop_conn()
1732 	if (con->sock && con->sock->sk)  in stop_conn()
1733 		con->sock->sk->sk_user_data = NULL;  in stop_conn()
1736 static void free_conn(struct connection *con)  in free_conn()  argument
1738 	close_connection(con, true);  in free_conn()
1739 	if (con->othercon)  in free_conn()
1740 		kmem_cache_free(con_cache, con->othercon);  in free_conn()
1741 	hlist_del(&con->list);  in free_conn()
1742 	kmem_cache_free(con_cache, con);  in free_conn()
1769 	struct connection *con;  in dlm_lowcomms_start()  local
1807 	con = nodeid2con(0,0);  in dlm_lowcomms_start()
1808 	if (con) {  in dlm_lowcomms_start()
1809 		close_connection(con, false);  in dlm_lowcomms_start()
1810 		kmem_cache_free(con_cache, con);  in dlm_lowcomms_start()