1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
4 * Copyright (C) 2004 Cluster File Systems, Inc.
5 * Author: Eric Barton <eric@bartonsoftware.com>
7 * This file is part of Lustre, http://www.lustre.org.
9 * Lustre is free software; you can redistribute it and/or
10 * modify it under the terms of version 2 of the GNU General Public
11 * License as published by the Free Software Foundation.
13 * Lustre is distributed in the hope that it will be useful,
14 * but WITHOUT ANY WARRANTY; without even the implied warranty of
15 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 * GNU General Public License for more details.
18 * You should have received a copy of the GNU General Public License
19 * along with Lustre; if not, write to the Free Software
20 * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
24 #include "openibnal.h"
27 ptl_handle_ni_t kibnal_ni;
28 kib_data_t kibnal_data;
29 kib_tunables_t kibnal_tunables;
31 #define IBNAL_SYSCTL 202
34 IBNAL_SYSCTL_TIMEOUT=1,
35 IBNAL_SYSCTL_LISTENER_TIMEOUT,
40 static ctl_table kibnal_ctl_table[] = {
41 {IBNAL_SYSCTL_TIMEOUT, "timeout",
42 &kibnal_tunables.kib_io_timeout, sizeof (int),
43 0644, NULL, &proc_dointvec},
44 {IBNAL_SYSCTL_LISTENER_TIMEOUT, "listener_timeout",
45 &kibnal_tunables.kib_listener_timeout, sizeof(int),
46 0644, NULL, &proc_dointvec},
47 {IBNAL_SYSCTL_BACKLOG, "backlog",
48 &kibnal_tunables.kib_backlog, sizeof(int),
49 0644, NULL, kibnal_listener_procint},
50 {IBNAL_SYSCTL_PORT, "port",
51 &kibnal_tunables.kib_port, sizeof(int),
52 0644, NULL, kibnal_listener_procint},
56 static ctl_table kibnal_top_ctl_table[] = {
57 {IBNAL_SYSCTL, "openibnal", NULL, 0, 0555, kibnal_ctl_table},
62 kibnal_cksum (void *ptr, int nob)
68 sum = ((sum << 1) | (sum >> 31)) + *c++;
70 /* ensure I don't return 0 (== no checksum) */
71 return (sum == 0) ? 1 : sum;
75 kibnal_init_msg(kib_msg_t *msg, int type, int body_nob)
78 msg->ibm_nob = offsetof(kib_msg_t, ibm_u) + body_nob;
82 kibnal_pack_msg(kib_msg_t *msg, int credits, ptl_nid_t dstnid, __u64 dststamp)
84 /* CAVEAT EMPTOR! all message fields not set here should have been
85 * initialised previously. */
86 msg->ibm_magic = IBNAL_MSG_MAGIC;
87 msg->ibm_version = IBNAL_MSG_VERSION;
89 msg->ibm_credits = credits;
92 msg->ibm_srcnid = kibnal_lib.libnal_ni.ni_pid.nid;
93 msg->ibm_srcstamp = kibnal_data.kib_incarnation;
94 msg->ibm_dstnid = dstnid;
95 msg->ibm_dststamp = dststamp;
97 /* NB ibm_cksum zero while computing cksum */
98 msg->ibm_cksum = kibnal_cksum(msg, msg->ibm_nob);
103 kibnal_unpack_msg(kib_msg_t *msg, int nob)
105 const int hdr_size = offsetof(kib_msg_t, ibm_u);
111 CERROR("Short message: %d\n", nob);
115 if (msg->ibm_magic == IBNAL_MSG_MAGIC) {
117 } else if (msg->ibm_magic == __swab32(IBNAL_MSG_MAGIC)) {
120 CERROR("Bad magic: %08x\n", msg->ibm_magic);
124 if (msg->ibm_version !=
125 (flip ? __swab16(IBNAL_MSG_VERSION) : IBNAL_MSG_VERSION)) {
126 CERROR("Bad version: %d\n", msg->ibm_version);
130 if (nob < hdr_size) {
131 CERROR("Short message: %d\n", nob);
135 msg_nob = flip ? __swab32(msg->ibm_nob) : msg->ibm_nob;
137 CERROR("Short message: got %d, wanted %d\n", nob, msg_nob);
141 /* checksum must be computed with ibm_cksum zero and BEFORE anything
143 msg_cksum = flip ? __swab32(msg->ibm_cksum) : msg->ibm_cksum;
145 if (msg_cksum != 0 &&
146 msg_cksum != kibnal_cksum(msg, msg_nob)) {
147 CERROR("Bad checksum\n");
150 msg->ibm_cksum = msg_cksum;
153 /* leave magic unflipped as a clue to peer endianness */
154 __swab16s(&msg->ibm_version);
155 LASSERT (sizeof(msg->ibm_type) == 1);
156 LASSERT (sizeof(msg->ibm_credits) == 1);
157 msg->ibm_nob = msg_nob;
158 __swab64s(&msg->ibm_srcnid);
159 __swab64s(&msg->ibm_srcstamp);
160 __swab64s(&msg->ibm_dstnid);
161 __swab64s(&msg->ibm_dststamp);
164 switch (msg->ibm_type) {
166 CERROR("Unknown message type %x\n", msg->ibm_type);
169 case IBNAL_MSG_SVCQRY:
173 case IBNAL_MSG_SVCRSP:
174 if (msg_nob < hdr_size + sizeof(msg->ibm_u.svcrsp)) {
175 CERROR("Short SVCRSP: %d(%d)\n", msg_nob,
176 (int)(hdr_size + sizeof(msg->ibm_u.svcrsp)));
180 __swab64s(&msg->ibm_u.svcrsp.ibsr_svc_id);
181 __swab16s(&msg->ibm_u.svcrsp.ibsr_svc_pkey);
185 case IBNAL_MSG_CONNREQ:
186 case IBNAL_MSG_CONNACK:
187 if (msg_nob < hdr_size + sizeof(msg->ibm_u.connparams)) {
188 CERROR("Short CONNREQ: %d(%d)\n", msg_nob,
189 (int)(hdr_size + sizeof(msg->ibm_u.connparams)));
193 __swab32s(&msg->ibm_u.connparams.ibcp_queue_depth);
196 case IBNAL_MSG_IMMEDIATE:
197 if (msg_nob < offsetof(kib_msg_t, ibm_u.immediate.ibim_payload[0])) {
198 CERROR("Short IMMEDIATE: %d(%d)\n", msg_nob,
199 (int)offsetof(kib_msg_t, ibm_u.immediate.ibim_payload[0]));
204 case IBNAL_MSG_PUT_RDMA:
205 case IBNAL_MSG_GET_RDMA:
206 if (msg_nob < hdr_size + sizeof(msg->ibm_u.rdma)) {
207 CERROR("Short RDMA req: %d(%d)\n", msg_nob,
208 (int)(hdr_size + sizeof(msg->ibm_u.rdma)));
212 __swab32s(&msg->ibm_u.rdma.ibrm_desc.rd_key);
213 __swab32s(&msg->ibm_u.rdma.ibrm_desc.rd_nob);
214 __swab64s(&msg->ibm_u.rdma.ibrm_desc.rd_addr);
218 case IBNAL_MSG_PUT_DONE:
219 case IBNAL_MSG_GET_DONE:
220 if (msg_nob < hdr_size + sizeof(msg->ibm_u.completion)) {
221 CERROR("Short RDMA completion: %d(%d)\n", msg_nob,
222 (int)(hdr_size + sizeof(msg->ibm_u.completion)));
226 __swab32s(&msg->ibm_u.completion.ibcm_status);
233 kibnal_sock_write (struct socket *sock, void *buffer, int nob)
236 mm_segment_t oldmm = get_fs();
241 struct msghdr msg = {
248 .msg_flags = MSG_DONTWAIT
251 /* We've set up the socket's send buffer to be large enough for
252 * everything we send, so a single non-blocking send should
253 * complete without error. */
256 rc = sock_sendmsg(sock, &msg, iov.iov_len);
269 kibnal_sock_read (struct socket *sock, void *buffer, int nob, int timeout)
272 mm_segment_t oldmm = get_fs();
273 long ticks = timeout * HZ;
285 struct msghdr msg = {
295 /* Set receive timeout to remaining time */
296 tv = (struct timeval) {
297 .tv_sec = ticks / HZ,
298 .tv_usec = ((ticks % HZ) * 1000000) / HZ
301 rc = sock_setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO,
302 (char *)&tv, sizeof(tv));
305 CERROR("Can't set socket recv timeout %d: %d\n",
312 rc = sock_recvmsg(sock, &msg, iov.iov_len, 0);
313 ticks -= jiffies - then;
320 return -ECONNABORTED;
322 buffer = ((char *)buffer) + rc;
334 kibnal_create_sock(struct socket **sockp)
339 mm_segment_t oldmm = get_fs();
341 rc = sock_create(PF_INET, SOCK_STREAM, 0, &sock);
343 CERROR("Can't create socket: %d\n", rc);
347 /* Ensure sends will not block */
348 option = 2 * sizeof(kib_msg_t);
350 rc = sock_setsockopt(sock, SOL_SOCKET, SO_SNDBUF,
351 (char *)&option, sizeof(option));
354 CERROR("Can't set send buffer %d: %d\n", option, rc);
360 rc = sock_setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
361 (char *)&option, sizeof(option));
364 CERROR("Can't set SO_REUSEADDR: %d\n", rc);
377 kibnal_pause(int ticks)
379 set_current_state(TASK_UNINTERRUPTIBLE);
380 schedule_timeout(ticks);
384 kibnal_connect_sock(kib_peer_t *peer, struct socket **sockp)
386 struct sockaddr_in locaddr;
387 struct sockaddr_in srvaddr;
392 for (port = 1023; port >= 512; port--) {
394 memset(&locaddr, 0, sizeof(locaddr));
395 locaddr.sin_family = AF_INET;
396 locaddr.sin_port = htons(port);
397 locaddr.sin_addr.s_addr = htonl(INADDR_ANY);
399 memset (&srvaddr, 0, sizeof (srvaddr));
400 srvaddr.sin_family = AF_INET;
401 srvaddr.sin_port = htons (peer->ibp_port);
402 srvaddr.sin_addr.s_addr = htonl (peer->ibp_ip);
404 rc = kibnal_create_sock(&sock);
408 rc = sock->ops->bind(sock,
409 (struct sockaddr *)&locaddr, sizeof(locaddr));
413 if (rc == -EADDRINUSE) {
414 CDEBUG(D_NET, "Port %d already in use\n", port);
418 CERROR("Can't bind to reserved port %d: %d\n", port, rc);
422 rc = sock->ops->connect(sock,
423 (struct sockaddr *)&srvaddr, sizeof(srvaddr),
432 if (rc != -EADDRNOTAVAIL) {
433 CERROR("Can't connect port %d to %u.%u.%u.%u/%d: %d\n",
434 port, HIPQUAD(peer->ibp_ip), peer->ibp_port, rc);
438 CDEBUG(D_NET, "Port %d not available for %u.%u.%u.%u/%d\n",
439 port, HIPQUAD(peer->ibp_ip), peer->ibp_port);
443 return -EHOSTUNREACH;
447 kibnal_make_svcqry (kib_conn_t *conn)
449 kib_peer_t *peer = conn->ibc_peer;
455 LASSERT (conn->ibc_connreq != NULL);
456 msg = &conn->ibc_connreq->cr_msg;
458 kibnal_init_msg(msg, IBNAL_MSG_SVCQRY, 0);
459 kibnal_pack_msg(msg, 0, peer->ibp_nid, 0);
461 rc = kibnal_connect_sock(peer, &sock);
465 rc = kibnal_sock_write(sock, msg, msg->ibm_nob);
467 CERROR("Error %d sending svcqry to "
468 LPX64"@%u.%u.%u.%u/%d\n", rc,
469 peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
473 nob = offsetof(kib_msg_t, ibm_u) + sizeof(msg->ibm_u.svcrsp);
474 rc = kibnal_sock_read(sock, msg, nob, kibnal_tunables.kib_io_timeout);
476 CERROR("Error %d receiving svcrsp from "
477 LPX64"@%u.%u.%u.%u/%d\n", rc,
478 peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
482 rc = kibnal_unpack_msg(msg, nob);
484 CERROR("Error %d unpacking svcrsp from "
485 LPX64"@%u.%u.%u.%u/%d\n", rc,
486 peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
490 if (msg->ibm_type != IBNAL_MSG_SVCRSP) {
491 CERROR("Unexpected response type %d from "
492 LPX64"@%u.%u.%u.%u/%d\n", msg->ibm_type,
493 peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
498 if (msg->ibm_dstnid != kibnal_lib.libnal_ni.ni_pid.nid ||
499 msg->ibm_dststamp != kibnal_data.kib_incarnation) {
500 CERROR("Unexpected dst NID/stamp "LPX64"/"LPX64" from "
501 LPX64"@%u.%u.%u.%u/%d\n",
502 msg->ibm_dstnid, msg->ibm_dststamp,
503 peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
508 if (msg->ibm_srcnid != peer->ibp_nid) {
509 CERROR("Unexpected src NID "LPX64" from "
510 LPX64"@%u.%u.%u.%u/%d\n", msg->ibm_srcnid,
511 peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
516 conn->ibc_incarnation = msg->ibm_srcstamp;
517 conn->ibc_connreq->cr_svcrsp = msg->ibm_u.svcrsp;
524 kibnal_handle_svcqry (struct socket *sock)
526 struct sockaddr_in addr;
528 unsigned int peer_port;
536 rc = sock->ops->getname(sock, (struct sockaddr *)&addr, &len, 2);
538 CERROR("Can't get peer's IP: %d\n", rc);
542 peer_ip = ntohl(addr.sin_addr.s_addr);
543 peer_port = ntohs(addr.sin_port);
545 if (peer_port >= 1024) {
546 CERROR("Refusing unprivileged connection from %u.%u.%u.%u/%d\n",
547 HIPQUAD(peer_ip), peer_port);
551 PORTAL_ALLOC(msg, sizeof(*msg));
553 CERROR("Can't allocate msgs for %u.%u.%u.%u/%d\n",
554 HIPQUAD(peer_ip), peer_port);
558 rc = kibnal_sock_read(sock, msg, offsetof(kib_msg_t, ibm_u),
559 kibnal_tunables.kib_listener_timeout);
561 CERROR("Error %d receiving svcqry from %u.%u.%u.%u/%d\n",
562 rc, HIPQUAD(peer_ip), peer_port);
566 rc = kibnal_unpack_msg(msg, offsetof(kib_msg_t, ibm_u));
568 CERROR("Error %d unpacking svcqry from %u.%u.%u.%u/%d\n",
569 rc, HIPQUAD(peer_ip), peer_port);
573 if (msg->ibm_type != IBNAL_MSG_SVCQRY) {
574 CERROR("Unexpected message %d from %u.%u.%u.%u/%d\n",
575 msg->ibm_type, HIPQUAD(peer_ip), peer_port);
579 if (msg->ibm_dstnid != kibnal_lib.libnal_ni.ni_pid.nid) {
580 CERROR("Unexpected dstnid "LPX64"(expected "LPX64" "
581 "from %u.%u.%u.%u/%d\n", msg->ibm_dstnid,
582 kibnal_lib.libnal_ni.ni_pid.nid,
583 HIPQUAD(peer_ip), peer_port);
587 srcnid = msg->ibm_srcnid;
588 srcstamp = msg->ibm_srcstamp;
590 kibnal_init_msg(msg, IBNAL_MSG_SVCRSP, sizeof(msg->ibm_u.svcrsp));
592 msg->ibm_u.svcrsp.ibsr_svc_id = kibnal_data.kib_svc_id;
593 memcpy(msg->ibm_u.svcrsp.ibsr_svc_gid, kibnal_data.kib_svc_gid,
594 sizeof(kibnal_data.kib_svc_gid));
595 msg->ibm_u.svcrsp.ibsr_svc_pkey = kibnal_data.kib_svc_pkey;
597 kibnal_pack_msg(msg, 0, srcnid, srcstamp);
599 rc = kibnal_sock_write (sock, msg, msg->ibm_nob);
601 CERROR("Error %d replying to svcqry from %u.%u.%u.%u/%d\n",
602 rc, HIPQUAD(peer_ip), peer_port);
607 PORTAL_FREE(msg, sizeof(*msg));
611 kibnal_free_acceptsock (kib_acceptsock_t *as)
613 sock_release(as->ibas_sock);
614 PORTAL_FREE(as, sizeof(*as));
618 kibnal_ip_listener(void *arg)
620 struct sockaddr_in addr;
623 kib_acceptsock_t *as;
629 /* Parent thread holds kib_nid_mutex, and is, or is about to
630 * block on kib_listener_signal */
632 port = kibnal_tunables.kib_port;
633 snprintf(name, sizeof(name), "kibnal_lstn%03d", port);
634 kportal_daemonize(name);
635 kportal_blockallsigs();
637 init_waitqueue_entry(&wait, current);
639 rc = kibnal_create_sock(&sock);
643 memset(&addr, 0, sizeof(addr));
644 addr.sin_family = AF_INET;
645 addr.sin_port = htons(port);
646 addr.sin_addr.s_addr = INADDR_ANY;
648 rc = sock->ops->bind(sock, (struct sockaddr *)&addr, sizeof(addr));
650 CERROR("Can't bind to port %d\n", port);
654 rc = sock->ops->listen(sock, kibnal_tunables.kib_backlog);
656 CERROR("Can't set listen backlog %d: %d\n",
657 kibnal_tunables.kib_backlog, rc);
661 LASSERT (kibnal_data.kib_listener_sock == NULL);
662 kibnal_data.kib_listener_sock = sock;
664 /* unblock waiting parent */
665 LASSERT (kibnal_data.kib_listener_shutdown == 0);
666 up(&kibnal_data.kib_listener_signal);
668 /* Wake me any time something happens on my socket */
669 add_wait_queue(sock->sk->sk_sleep, &wait);
672 while (kibnal_data.kib_listener_shutdown == 0) {
675 PORTAL_ALLOC(as, sizeof(*as));
677 CERROR("Out of Memory: pausing...\n");
681 as->ibas_sock = NULL;
684 if (as->ibas_sock == NULL) {
685 as->ibas_sock = sock_alloc();
686 if (as->ibas_sock == NULL) {
687 CERROR("Can't allocate socket: pausing...\n");
691 /* XXX this should add a ref to sock->ops->owner, if
692 * TCP could be a module */
693 as->ibas_sock->type = sock->type;
694 as->ibas_sock->ops = sock->ops;
697 set_current_state(TASK_INTERRUPTIBLE);
699 rc = sock->ops->accept(sock, as->ibas_sock, O_NONBLOCK);
701 /* Sleep for socket activity? */
703 kibnal_data.kib_listener_shutdown == 0)
706 set_current_state(TASK_RUNNING);
709 spin_lock_irqsave(&kibnal_data.kib_connd_lock, flags);
711 list_add_tail(&as->ibas_list,
712 &kibnal_data.kib_connd_acceptq);
714 spin_unlock_irqrestore(&kibnal_data.kib_connd_lock, flags);
715 wake_up(&kibnal_data.kib_connd_waitq);
722 CERROR("Accept failed: %d, pausing...\n", rc);
728 if (as->ibas_sock != NULL)
729 sock_release(as->ibas_sock);
730 PORTAL_FREE(as, sizeof(*as));
734 remove_wait_queue(sock->sk->sk_sleep, &wait);
737 kibnal_data.kib_listener_sock = NULL;
739 /* set completion status and unblock thread waiting for me
740 * (parent on startup failure, executioner on normal shutdown) */
741 kibnal_data.kib_listener_shutdown = rc;
742 up(&kibnal_data.kib_listener_signal);
748 kibnal_start_ip_listener (void)
753 CDEBUG(D_NET, "Starting listener\n");
755 /* Called holding kib_nid_mutex: listener stopped */
756 LASSERT (kibnal_data.kib_listener_sock == NULL);
758 kibnal_data.kib_listener_shutdown = 0;
759 pid = kernel_thread(kibnal_ip_listener, NULL, 0);
761 CERROR("Can't spawn listener: %ld\n", pid);
765 /* Block until listener has started up. */
766 down(&kibnal_data.kib_listener_signal);
768 rc = kibnal_data.kib_listener_shutdown;
769 LASSERT ((rc != 0) == (kibnal_data.kib_listener_sock == NULL));
771 CDEBUG((rc == 0) ? D_WARNING : D_ERROR,
772 "Listener %s: pid:%ld port:%d backlog:%d\n",
773 (rc == 0) ? "started OK" : "startup failed",
774 pid, kibnal_tunables.kib_port, kibnal_tunables.kib_backlog);
780 kibnal_stop_ip_listener(int clear_acceptq)
782 struct list_head zombie_accepts;
783 kib_acceptsock_t *as;
786 CDEBUG(D_NET, "Stopping listener\n");
788 /* Called holding kib_nid_mutex: listener running */
789 LASSERT (kibnal_data.kib_listener_sock != NULL);
791 kibnal_data.kib_listener_shutdown = 1;
792 wake_up_all(kibnal_data.kib_listener_sock->sk->sk_sleep);
794 /* Block until listener has torn down. */
795 down(&kibnal_data.kib_listener_signal);
797 LASSERT (kibnal_data.kib_listener_sock == NULL);
798 CDEBUG(D_WARNING, "Listener stopped\n");
803 /* Close any unhandled accepts */
804 spin_lock_irqsave(&kibnal_data.kib_connd_lock, flags);
806 list_add(&zombie_accepts, &kibnal_data.kib_connd_acceptq);
807 list_del_init(&kibnal_data.kib_connd_acceptq);
809 spin_unlock_irqrestore(&kibnal_data.kib_connd_lock, flags);
811 while (!list_empty(&zombie_accepts)) {
812 as = list_entry(zombie_accepts.next,
813 kib_acceptsock_t, ibas_list);
814 list_del(&as->ibas_list);
815 kibnal_free_acceptsock(as);
820 kibnal_listener_procint(ctl_table *table, int write, struct file *filp,
821 void *buffer, size_t *lenp)
823 int *tunable = (int *)table->data;
827 /* No race with nal initialisation since the nal is setup all the time
828 * it's loaded. When that changes, change this! */
829 LASSERT (kibnal_data.kib_init == IBNAL_INIT_ALL);
831 down(&kibnal_data.kib_nid_mutex);
833 LASSERT (tunable == &kibnal_tunables.kib_port ||
834 tunable == &kibnal_tunables.kib_backlog);
837 rc = proc_dointvec(table, write, filp, buffer, lenp);
840 (*tunable != old_val ||
841 kibnal_data.kib_listener_sock == NULL)) {
843 if (kibnal_data.kib_listener_sock != NULL)
844 kibnal_stop_ip_listener(0);
846 rc = kibnal_start_ip_listener();
848 CERROR("Unable to restart listener with new tunable:"
849 " reverting to old value\n");
851 kibnal_start_ip_listener();
855 up(&kibnal_data.kib_nid_mutex);
857 LASSERT (kibnal_data.kib_init == IBNAL_INIT_ALL);
862 kibnal_start_ib_listener (void)
866 LASSERT (kibnal_data.kib_listen_handle == NULL);
868 kibnal_data.kib_svc_id = ib_cm_service_assign();
869 CDEBUG(D_NET, "svc id "LPX64"\n", kibnal_data.kib_svc_id);
871 rc = ib_cached_gid_get(kibnal_data.kib_device,
872 kibnal_data.kib_port, 0,
873 kibnal_data.kib_svc_gid);
875 CERROR("Can't get port %d GID: %d\n",
876 kibnal_data.kib_port, rc);
880 rc = ib_cached_pkey_get(kibnal_data.kib_device,
881 kibnal_data.kib_port, 0,
882 &kibnal_data.kib_svc_pkey);
884 CERROR ("Can't get port %d PKEY: %d\n",
885 kibnal_data.kib_port, rc);
889 rc = ib_cm_listen(kibnal_data.kib_svc_id,
890 TS_IB_CM_SERVICE_EXACT_MASK,
891 kibnal_passive_conn_callback, NULL,
892 &kibnal_data.kib_listen_handle);
894 kibnal_data.kib_listen_handle = NULL;
895 CERROR ("Can't create IB listener: %d\n", rc);
899 LASSERT (kibnal_data.kib_listen_handle != NULL);
904 kibnal_stop_ib_listener (void)
908 LASSERT (kibnal_data.kib_listen_handle != NULL);
910 rc = ib_cm_listen_stop (kibnal_data.kib_listen_handle);
912 CERROR("Error stopping IB listener: %d\n", rc);
914 kibnal_data.kib_listen_handle = NULL;
918 kibnal_set_mynid (ptl_nid_t nid)
920 lib_ni_t *ni = &kibnal_lib.libnal_ni;
923 CDEBUG(D_IOCTL, "setting mynid to "LPX64" (old nid="LPX64")\n",
924 nid, ni->ni_pid.nid);
926 down (&kibnal_data.kib_nid_mutex);
928 if (nid == kibnal_data.kib_nid) {
929 /* no change of NID */
930 up (&kibnal_data.kib_nid_mutex);
934 CDEBUG(D_NET, "NID "LPX64"("LPX64")\n",
935 kibnal_data.kib_nid, nid);
937 if (kibnal_data.kib_listener_sock != NULL)
938 kibnal_stop_ip_listener(1);
940 if (kibnal_data.kib_listen_handle != NULL)
941 kibnal_stop_ib_listener();
943 ni->ni_pid.nid = nid;
944 kibnal_data.kib_incarnation++;
946 /* Delete all existing peers and their connections after new
947 * NID/incarnation set to ensure no old connections in our brave new
949 kibnal_del_peer (PTL_NID_ANY, 0);
951 if (ni->ni_pid.nid != PTL_NID_ANY) {
952 /* got a new NID to install */
953 rc = kibnal_start_ib_listener();
955 CERROR("Can't start IB listener: %d\n", rc);
959 rc = kibnal_start_ip_listener();
961 CERROR("Can't start IP listener: %d\n", rc);
966 up(&kibnal_data.kib_nid_mutex);
970 kibnal_stop_ib_listener();
972 ni->ni_pid.nid = PTL_NID_ANY;
973 kibnal_data.kib_incarnation++;
975 kibnal_del_peer (PTL_NID_ANY, 0);
976 up(&kibnal_data.kib_nid_mutex);
981 kibnal_create_peer (ptl_nid_t nid)
985 LASSERT (nid != PTL_NID_ANY);
987 PORTAL_ALLOC (peer, sizeof (*peer));
991 memset(peer, 0, sizeof(*peer)); /* zero flags etc */
994 atomic_set (&peer->ibp_refcount, 1); /* 1 ref for caller */
996 INIT_LIST_HEAD (&peer->ibp_list); /* not in the peer table yet */
997 INIT_LIST_HEAD (&peer->ibp_conns);
998 INIT_LIST_HEAD (&peer->ibp_tx_queue);
1000 peer->ibp_reconnect_time = jiffies;
1001 peer->ibp_reconnect_interval = IBNAL_MIN_RECONNECT_INTERVAL;
1003 atomic_inc (&kibnal_data.kib_npeers);
1004 CDEBUG(D_NET, "peer %p "LPX64"\n", peer, nid);
1010 kibnal_destroy_peer (kib_peer_t *peer)
1012 CDEBUG (D_NET, "peer "LPX64" %p deleted\n", peer->ibp_nid, peer);
1014 LASSERT (atomic_read (&peer->ibp_refcount) == 0);
1015 LASSERT (peer->ibp_persistence == 0);
1016 LASSERT (!kibnal_peer_active(peer));
1017 LASSERT (peer->ibp_connecting == 0);
1018 LASSERT (list_empty (&peer->ibp_conns));
1019 LASSERT (list_empty (&peer->ibp_tx_queue));
1021 PORTAL_FREE (peer, sizeof (*peer));
1023 /* NB a peer's connections keep a reference on their peer until
1024 * they are destroyed, so we can be assured that _all_ state to do
1025 * with this peer has been cleaned up when its refcount drops to
1027 atomic_dec (&kibnal_data.kib_npeers);
1031 kibnal_put_peer (kib_peer_t *peer)
1033 CDEBUG (D_OTHER, "putting peer[%p] -> "LPX64" (%d)\n",
1034 peer, peer->ibp_nid,
1035 atomic_read (&peer->ibp_refcount));
1037 LASSERT (atomic_read (&peer->ibp_refcount) > 0);
1038 if (!atomic_dec_and_test (&peer->ibp_refcount))
1041 kibnal_destroy_peer (peer);
1045 kibnal_find_peer_locked (ptl_nid_t nid)
1047 struct list_head *peer_list = kibnal_nid2peerlist (nid);
1048 struct list_head *tmp;
1051 list_for_each (tmp, peer_list) {
1053 peer = list_entry (tmp, kib_peer_t, ibp_list);
1055 LASSERT (peer->ibp_persistence != 0 || /* persistent peer */
1056 peer->ibp_connecting != 0 || /* creating conns */
1057 !list_empty (&peer->ibp_conns)); /* active conn */
1059 if (peer->ibp_nid != nid)
1062 CDEBUG(D_NET, "got peer [%p] -> "LPX64" (%d)\n",
1063 peer, nid, atomic_read (&peer->ibp_refcount));
1070 kibnal_get_peer (ptl_nid_t nid)
1074 read_lock (&kibnal_data.kib_global_lock);
1075 peer = kibnal_find_peer_locked (nid);
1076 if (peer != NULL) /* +1 ref for caller? */
1077 atomic_inc (&peer->ibp_refcount);
1078 read_unlock (&kibnal_data.kib_global_lock);
1084 kibnal_unlink_peer_locked (kib_peer_t *peer)
1086 LASSERT (peer->ibp_persistence == 0);
1087 LASSERT (list_empty(&peer->ibp_conns));
1089 LASSERT (kibnal_peer_active(peer));
1090 list_del_init (&peer->ibp_list);
1091 /* lose peerlist's ref */
1092 kibnal_put_peer (peer);
1096 kibnal_get_peer_info (int index, ptl_nid_t *nidp, __u32 *ipp, int *portp,
1100 struct list_head *ptmp;
1103 read_lock (&kibnal_data.kib_global_lock);
1105 for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
1107 list_for_each (ptmp, &kibnal_data.kib_peers[i]) {
1109 peer = list_entry (ptmp, kib_peer_t, ibp_list);
1110 LASSERT (peer->ibp_persistence != 0 ||
1111 peer->ibp_connecting != 0 ||
1112 !list_empty (&peer->ibp_conns));
1117 *nidp = peer->ibp_nid;
1118 *ipp = peer->ibp_ip;
1119 *portp = peer->ibp_port;
1120 *persistencep = peer->ibp_persistence;
1122 read_unlock (&kibnal_data.kib_global_lock);
1127 read_unlock (&kibnal_data.kib_global_lock);
1132 kibnal_add_persistent_peer (ptl_nid_t nid, __u32 ip, int port)
1134 unsigned long flags;
1138 if (nid == PTL_NID_ANY)
1141 peer = kibnal_create_peer (nid);
1145 write_lock_irqsave (&kibnal_data.kib_global_lock, flags);
1147 peer2 = kibnal_find_peer_locked (nid);
1148 if (peer2 != NULL) {
1149 kibnal_put_peer (peer);
1152 /* peer table takes existing ref on peer */
1153 list_add_tail (&peer->ibp_list,
1154 kibnal_nid2peerlist (nid));
1158 peer->ibp_port = port;
1159 peer->ibp_persistence++;
1161 write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
1166 kibnal_del_peer_locked (kib_peer_t *peer, int single_share)
1168 struct list_head *ctmp;
1169 struct list_head *cnxt;
1173 peer->ibp_persistence = 0;
1174 else if (peer->ibp_persistence > 0)
1175 peer->ibp_persistence--;
1177 if (peer->ibp_persistence != 0)
1180 if (list_empty(&peer->ibp_conns)) {
1181 kibnal_unlink_peer_locked(peer);
1183 list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
1184 conn = list_entry(ctmp, kib_conn_t, ibc_list);
1186 kibnal_close_conn_locked (conn, 0);
1188 /* NB peer is no longer persistent; closing its last conn
1191 /* NB peer now unlinked; might even be freed if the peer table had the
1192 * last ref on it. */
1196 kibnal_del_peer (ptl_nid_t nid, int single_share)
1198 unsigned long flags;
1199 struct list_head *ptmp;
1200 struct list_head *pnxt;
1207 write_lock_irqsave (&kibnal_data.kib_global_lock, flags);
1209 if (nid != PTL_NID_ANY)
1210 lo = hi = kibnal_nid2peerlist(nid) - kibnal_data.kib_peers;
1213 hi = kibnal_data.kib_peer_hash_size - 1;
1216 for (i = lo; i <= hi; i++) {
1217 list_for_each_safe (ptmp, pnxt, &kibnal_data.kib_peers[i]) {
1218 peer = list_entry (ptmp, kib_peer_t, ibp_list);
1219 LASSERT (peer->ibp_persistence != 0 ||
1220 peer->ibp_connecting != 0 ||
1221 !list_empty (&peer->ibp_conns));
1223 if (!(nid == PTL_NID_ANY || peer->ibp_nid == nid))
1226 kibnal_del_peer_locked (peer, single_share);
1227 rc = 0; /* matched something */
1234 write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
1240 kibnal_get_conn_by_idx (int index)
1243 struct list_head *ptmp;
1245 struct list_head *ctmp;
1248 read_lock (&kibnal_data.kib_global_lock);
1250 for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
1251 list_for_each (ptmp, &kibnal_data.kib_peers[i]) {
1253 peer = list_entry (ptmp, kib_peer_t, ibp_list);
1254 LASSERT (peer->ibp_persistence > 0 ||
1255 peer->ibp_connecting != 0 ||
1256 !list_empty (&peer->ibp_conns));
1258 list_for_each (ctmp, &peer->ibp_conns) {
1262 conn = list_entry (ctmp, kib_conn_t, ibc_list);
1263 CDEBUG(D_NET, "++conn[%p] state %d -> "LPX64" (%d)\n",
1264 conn, conn->ibc_state, conn->ibc_peer->ibp_nid,
1265 atomic_read (&conn->ibc_refcount));
1266 atomic_inc (&conn->ibc_refcount);
1267 read_unlock (&kibnal_data.kib_global_lock);
1273 read_unlock (&kibnal_data.kib_global_lock);
1278 kibnal_create_conn (void)
1288 struct ib_qp_create_param qp_create;
1289 struct ib_qp_attribute qp_attr;
1292 PORTAL_ALLOC (conn, sizeof (*conn));
1294 CERROR ("Can't allocate connection\n");
1298 /* zero flags, NULL pointers etc... */
1299 memset (conn, 0, sizeof (*conn));
1301 INIT_LIST_HEAD (&conn->ibc_tx_queue);
1302 INIT_LIST_HEAD (&conn->ibc_active_txs);
1303 spin_lock_init (&conn->ibc_lock);
1305 atomic_inc (&kibnal_data.kib_nconns);
1306 /* well not really, but I call destroy() on failure, which decrements */
1308 PORTAL_ALLOC (conn->ibc_rxs, IBNAL_RX_MSGS * sizeof (kib_rx_t));
1309 if (conn->ibc_rxs == NULL)
1311 memset (conn->ibc_rxs, 0, IBNAL_RX_MSGS * sizeof(kib_rx_t));
1313 rc = kibnal_alloc_pages(&conn->ibc_rx_pages,
1315 IB_ACCESS_LOCAL_WRITE);
1319 vaddr_base = vaddr = conn->ibc_rx_pages->ibp_vaddr;
1321 for (i = ipage = page_offset = 0; i < IBNAL_RX_MSGS; i++) {
1322 struct page *page = conn->ibc_rx_pages->ibp_pages[ipage];
1323 kib_rx_t *rx = &conn->ibc_rxs[i];
1326 rx->rx_vaddr = vaddr;
1327 rx->rx_msg = (kib_msg_t *)(((char *)page_address(page)) + page_offset);
1329 vaddr += IBNAL_MSG_SIZE;
1330 LASSERT (vaddr <= vaddr_base + IBNAL_RX_MSG_BYTES);
1332 page_offset += IBNAL_MSG_SIZE;
1333 LASSERT (page_offset <= PAGE_SIZE);
1335 if (page_offset == PAGE_SIZE) {
1338 LASSERT (ipage <= IBNAL_RX_MSG_PAGES);
1342 params.qp_create = (struct ib_qp_create_param) {
1344 /* Sends have an optional RDMA */
1345 .max_outstanding_send_request = 2 * IBNAL_MSG_QUEUE_SIZE,
1346 .max_outstanding_receive_request = IBNAL_MSG_QUEUE_SIZE,
1347 .max_send_gather_element = 1,
1348 .max_receive_scatter_element = 1,
1350 .pd = kibnal_data.kib_pd,
1351 .send_queue = kibnal_data.kib_cq,
1352 .receive_queue = kibnal_data.kib_cq,
1353 .send_policy = IB_WQ_SIGNAL_SELECTABLE,
1354 .receive_policy = IB_WQ_SIGNAL_SELECTABLE,
1356 .transport = IB_TRANSPORT_RC,
1357 .device_specific = NULL,
1360 rc = ib_qp_create (¶ms.qp_create, &conn->ibc_qp, &conn->ibc_qpn);
1362 CERROR ("Failed to create queue pair: %d\n", rc);
1366 /* Mark QP created */
1367 conn->ibc_state = IBNAL_CONN_INIT_QP;
1369 params.qp_attr = (struct ib_qp_attribute) {
1370 .state = IB_QP_STATE_INIT,
1371 .port = kibnal_data.kib_port,
1372 .enable_rdma_read = 1,
1373 .enable_rdma_write = 1,
1374 .valid_fields = (IB_QP_ATTRIBUTE_STATE |
1375 IB_QP_ATTRIBUTE_PORT |
1376 IB_QP_ATTRIBUTE_PKEY_INDEX |
1377 IB_QP_ATTRIBUTE_RDMA_ATOMIC_ENABLE),
1379 rc = ib_qp_modify(conn->ibc_qp, ¶ms.qp_attr);
1381 CERROR ("Failed to modify queue pair: %d\n", rc);
1385 /* 1 ref for caller */
1386 atomic_set (&conn->ibc_refcount, 1);
1390 kibnal_destroy_conn (conn);
1395 kibnal_destroy_conn (kib_conn_t *conn)
1399 CDEBUG (D_NET, "connection %p\n", conn);
1401 LASSERT (atomic_read (&conn->ibc_refcount) == 0);
1402 LASSERT (list_empty(&conn->ibc_tx_queue));
1403 LASSERT (list_empty(&conn->ibc_active_txs));
1404 LASSERT (conn->ibc_nsends_posted == 0);
1405 LASSERT (conn->ibc_connreq == NULL);
1407 switch (conn->ibc_state) {
1408 case IBNAL_CONN_ZOMBIE:
1409 /* called after connection sequence initiated */
1411 case IBNAL_CONN_INIT_QP:
1412 rc = ib_qp_destroy(conn->ibc_qp);
1414 CERROR("Can't destroy QP: %d\n", rc);
1417 case IBNAL_CONN_INIT_NOTHING:
1424 if (conn->ibc_rx_pages != NULL)
1425 kibnal_free_pages(conn->ibc_rx_pages);
1427 if (conn->ibc_rxs != NULL)
1428 PORTAL_FREE(conn->ibc_rxs,
1429 IBNAL_RX_MSGS * sizeof(kib_rx_t));
1431 if (conn->ibc_peer != NULL)
1432 kibnal_put_peer(conn->ibc_peer);
1434 PORTAL_FREE(conn, sizeof (*conn));
1436 atomic_dec(&kibnal_data.kib_nconns);
1438 if (atomic_read (&kibnal_data.kib_nconns) == 0 &&
1439 kibnal_data.kib_shutdown) {
1440 /* I just nuked the last connection on shutdown; wake up
1441 * everyone so they can exit. */
1442 wake_up_all(&kibnal_data.kib_sched_waitq);
1443 wake_up_all(&kibnal_data.kib_reaper_waitq);
1448 kibnal_put_conn (kib_conn_t *conn)
1450 unsigned long flags;
1452 CDEBUG (D_NET, "putting conn[%p] state %d -> "LPX64" (%d)\n",
1453 conn, conn->ibc_state, conn->ibc_peer->ibp_nid,
1454 atomic_read (&conn->ibc_refcount));
1456 LASSERT (atomic_read (&conn->ibc_refcount) > 0);
1457 if (!atomic_dec_and_test (&conn->ibc_refcount))
1460 /* last ref only goes on zombies */
1461 LASSERT (conn->ibc_state == IBNAL_CONN_ZOMBIE);
1463 spin_lock_irqsave (&kibnal_data.kib_reaper_lock, flags);
1465 list_add (&conn->ibc_list, &kibnal_data.kib_reaper_conns);
1466 wake_up (&kibnal_data.kib_reaper_waitq);
1468 spin_unlock_irqrestore (&kibnal_data.kib_reaper_lock, flags);
1472 kibnal_close_peer_conns_locked (kib_peer_t *peer, int why)
1475 struct list_head *ctmp;
1476 struct list_head *cnxt;
1479 list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
1480 conn = list_entry (ctmp, kib_conn_t, ibc_list);
1483 kibnal_close_conn_locked (conn, why);
1490 kibnal_close_stale_conns_locked (kib_peer_t *peer, __u64 incarnation)
1493 struct list_head *ctmp;
1494 struct list_head *cnxt;
1497 list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
1498 conn = list_entry (ctmp, kib_conn_t, ibc_list);
1500 if (conn->ibc_incarnation == incarnation)
1503 CDEBUG(D_NET, "Closing stale conn nid:"LPX64" incarnation:"LPX64"("LPX64")\n",
1504 peer->ibp_nid, conn->ibc_incarnation, incarnation);
1507 kibnal_close_conn_locked (conn, -ESTALE);
1514 kibnal_close_matching_conns (ptl_nid_t nid)
1516 unsigned long flags;
1518 struct list_head *ptmp;
1519 struct list_head *pnxt;
1525 write_lock_irqsave (&kibnal_data.kib_global_lock, flags);
1527 if (nid != PTL_NID_ANY)
1528 lo = hi = kibnal_nid2peerlist(nid) - kibnal_data.kib_peers;
1531 hi = kibnal_data.kib_peer_hash_size - 1;
1534 for (i = lo; i <= hi; i++) {
1535 list_for_each_safe (ptmp, pnxt, &kibnal_data.kib_peers[i]) {
1537 peer = list_entry (ptmp, kib_peer_t, ibp_list);
1538 LASSERT (peer->ibp_persistence != 0 ||
1539 peer->ibp_connecting != 0 ||
1540 !list_empty (&peer->ibp_conns));
1542 if (!(nid == PTL_NID_ANY || nid == peer->ibp_nid))
1545 count += kibnal_close_peer_conns_locked (peer, 0);
1549 write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
1551 /* wildcards always succeed */
1552 if (nid == PTL_NID_ANY)
1555 return (count == 0 ? -ENOENT : 0);
1559 kibnal_cmd(struct portals_cfg *pcfg, void * private)
1563 LASSERT (pcfg != NULL);
1565 switch(pcfg->pcfg_command) {
1566 case NAL_CMD_GET_PEER: {
1570 int share_count = 0;
1572 rc = kibnal_get_peer_info(pcfg->pcfg_count,
1573 &nid, &ip, &port, &share_count);
1574 pcfg->pcfg_nid = nid;
1575 pcfg->pcfg_size = 0;
1577 pcfg->pcfg_misc = port;
1578 pcfg->pcfg_count = 0;
1579 pcfg->pcfg_wait = share_count;
1582 case NAL_CMD_ADD_PEER: {
1583 rc = kibnal_add_persistent_peer (pcfg->pcfg_nid,
1584 pcfg->pcfg_id, /* IP */
1585 pcfg->pcfg_misc); /* port */
1588 case NAL_CMD_DEL_PEER: {
1589 rc = kibnal_del_peer (pcfg->pcfg_nid,
1590 /* flags == single_share */
1591 pcfg->pcfg_flags != 0);
1594 case NAL_CMD_GET_CONN: {
1595 kib_conn_t *conn = kibnal_get_conn_by_idx (pcfg->pcfg_count);
1601 pcfg->pcfg_nid = conn->ibc_peer->ibp_nid;
1603 pcfg->pcfg_misc = 0;
1604 pcfg->pcfg_flags = 0;
1605 kibnal_put_conn (conn);
1609 case NAL_CMD_CLOSE_CONNECTION: {
1610 rc = kibnal_close_matching_conns (pcfg->pcfg_nid);
1613 case NAL_CMD_REGISTER_MYNID: {
1614 if (pcfg->pcfg_nid == PTL_NID_ANY)
1617 rc = kibnal_set_mynid (pcfg->pcfg_nid);
1626 kibnal_free_pages (kib_pages_t *p)
1628 int npages = p->ibp_npages;
1632 if (p->ibp_mapped) {
1633 rc = ib_memory_deregister(p->ibp_handle);
1635 CERROR ("Deregister error: %d\n", rc);
1638 for (i = 0; i < npages; i++)
1639 if (p->ibp_pages[i] != NULL)
1640 __free_page(p->ibp_pages[i]);
1642 PORTAL_FREE (p, offsetof(kib_pages_t, ibp_pages[npages]));
1646 kibnal_alloc_pages (kib_pages_t **pp, int npages, int access)
1649 struct ib_physical_buffer *phys_pages;
1653 PORTAL_ALLOC(p, offsetof(kib_pages_t, ibp_pages[npages]));
1655 CERROR ("Can't allocate buffer %d\n", npages);
1659 memset (p, 0, offsetof(kib_pages_t, ibp_pages[npages]));
1660 p->ibp_npages = npages;
1662 for (i = 0; i < npages; i++) {
1663 p->ibp_pages[i] = alloc_page (GFP_KERNEL);
1664 if (p->ibp_pages[i] == NULL) {
1665 CERROR ("Can't allocate page %d of %d\n", i, npages);
1666 kibnal_free_pages(p);
1671 PORTAL_ALLOC(phys_pages, npages * sizeof(*phys_pages));
1672 if (phys_pages == NULL) {
1673 CERROR ("Can't allocate physarray for %d pages\n", npages);
1674 kibnal_free_pages(p);
1678 for (i = 0; i < npages; i++) {
1679 phys_pages[i].size = PAGE_SIZE;
1680 phys_pages[i].address =
1681 kibnal_page2phys(p->ibp_pages[i]);
1685 rc = ib_memory_register_physical(kibnal_data.kib_pd,
1688 npages * PAGE_SIZE, 0,
1694 PORTAL_FREE(phys_pages, npages * sizeof(*phys_pages));
1697 CERROR ("Error %d mapping %d pages\n", rc, npages);
1698 kibnal_free_pages(p);
1708 kibnal_setup_tx_descs (void)
1711 int page_offset = 0;
1719 /* pre-mapped messages are not bigger than 1 page */
1720 LASSERT (IBNAL_MSG_SIZE <= PAGE_SIZE);
1722 /* No fancy arithmetic when we do the buffer calculations */
1723 LASSERT (PAGE_SIZE % IBNAL_MSG_SIZE == 0);
1725 rc = kibnal_alloc_pages(&kibnal_data.kib_tx_pages,
1727 0); /* local read access only */
1731 vaddr = vaddr_base = kibnal_data.kib_tx_pages->ibp_vaddr;
1733 for (i = 0; i < IBNAL_TX_MSGS; i++) {
1734 page = kibnal_data.kib_tx_pages->ibp_pages[ipage];
1735 tx = &kibnal_data.kib_tx_descs[i];
1737 memset (tx, 0, sizeof(*tx)); /* zero flags etc */
1739 tx->tx_msg = (kib_msg_t *)(((char *)page_address(page)) + page_offset);
1740 tx->tx_vaddr = vaddr;
1741 tx->tx_isnblk = (i >= IBNAL_NTX);
1742 tx->tx_mapped = KIB_TX_UNMAPPED;
1744 CDEBUG(D_NET, "Tx[%d] %p->%p - "LPX64"\n",
1745 i, tx, tx->tx_msg, tx->tx_vaddr);
1748 list_add (&tx->tx_list,
1749 &kibnal_data.kib_idle_nblk_txs);
1751 list_add (&tx->tx_list,
1752 &kibnal_data.kib_idle_txs);
1754 vaddr += IBNAL_MSG_SIZE;
1755 LASSERT (vaddr <= vaddr_base + IBNAL_TX_MSG_BYTES);
1757 page_offset += IBNAL_MSG_SIZE;
1758 LASSERT (page_offset <= PAGE_SIZE);
1760 if (page_offset == PAGE_SIZE) {
1763 LASSERT (ipage <= IBNAL_TX_MSG_PAGES);
1771 kibnal_api_shutdown (nal_t *nal)
1776 if (nal->nal_refct != 0) {
1777 /* This module got the first ref */
1778 PORTAL_MODULE_UNUSE;
1782 CDEBUG(D_MALLOC, "before NAL cleanup: kmem %d\n",
1783 atomic_read (&portal_kmemory));
1785 LASSERT(nal == &kibnal_api);
1787 switch (kibnal_data.kib_init) {
1789 CERROR ("Unexpected state %d\n", kibnal_data.kib_init);
1792 case IBNAL_INIT_ALL:
1793 /* stop calls to nal_cmd */
1794 libcfs_nal_cmd_unregister(OPENIBNAL);
1797 /* resetting my NID unadvertises me, removes my
1798 * listener and nukes all current peers */
1799 kibnal_set_mynid (PTL_NID_ANY);
1801 /* Wait for all peer state to clean up */
1803 while (atomic_read (&kibnal_data.kib_npeers) != 0) {
1805 CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1806 "waiting for %d peers to close down\n",
1807 atomic_read (&kibnal_data.kib_npeers));
1808 set_current_state (TASK_INTERRUPTIBLE);
1809 schedule_timeout (HZ);
1814 rc = ib_cq_destroy (kibnal_data.kib_cq);
1816 CERROR ("Destroy CQ error: %d\n", rc);
1819 case IBNAL_INIT_TXD:
1820 kibnal_free_pages (kibnal_data.kib_tx_pages);
1823 case IBNAL_INIT_FMR:
1824 rc = ib_fmr_pool_destroy (kibnal_data.kib_fmr_pool);
1826 CERROR ("Destroy FMR pool error: %d\n", rc);
1830 rc = ib_pd_destroy(kibnal_data.kib_pd);
1832 CERROR ("Destroy PD error: %d\n", rc);
1835 case IBNAL_INIT_LIB:
1836 lib_fini(&kibnal_lib);
1839 case IBNAL_INIT_DATA:
1840 /* Module refcount only gets to zero when all peers
1841 * have been closed so all lists must be empty */
1842 LASSERT (atomic_read (&kibnal_data.kib_npeers) == 0);
1843 LASSERT (kibnal_data.kib_peers != NULL);
1844 for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
1845 LASSERT (list_empty (&kibnal_data.kib_peers[i]));
1847 LASSERT (atomic_read (&kibnal_data.kib_nconns) == 0);
1848 LASSERT (list_empty (&kibnal_data.kib_sched_rxq));
1849 LASSERT (list_empty (&kibnal_data.kib_sched_txq));
1850 LASSERT (list_empty (&kibnal_data.kib_reaper_conns));
1851 LASSERT (list_empty (&kibnal_data.kib_connd_peers));
1852 LASSERT (list_empty (&kibnal_data.kib_connd_acceptq));
1854 /* flag threads to terminate; wake and wait for them to die */
1855 kibnal_data.kib_shutdown = 1;
1856 wake_up_all (&kibnal_data.kib_sched_waitq);
1857 wake_up_all (&kibnal_data.kib_reaper_waitq);
1858 wake_up_all (&kibnal_data.kib_connd_waitq);
1861 while (atomic_read (&kibnal_data.kib_nthreads) != 0) {
1863 CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1864 "Waiting for %d threads to terminate\n",
1865 atomic_read (&kibnal_data.kib_nthreads));
1866 set_current_state (TASK_INTERRUPTIBLE);
1867 schedule_timeout (HZ);
1871 case IBNAL_INIT_NOTHING:
1875 if (kibnal_data.kib_tx_descs != NULL)
1876 PORTAL_FREE (kibnal_data.kib_tx_descs,
1877 IBNAL_TX_MSGS * sizeof(kib_tx_t));
1879 if (kibnal_data.kib_peers != NULL)
1880 PORTAL_FREE (kibnal_data.kib_peers,
1881 sizeof (struct list_head) *
1882 kibnal_data.kib_peer_hash_size);
1884 CDEBUG(D_MALLOC, "after NAL cleanup: kmem %d\n",
1885 atomic_read (&portal_kmemory));
1886 printk(KERN_INFO "Lustre: OpenIB NAL unloaded (final mem %d)\n",
1887 atomic_read(&portal_kmemory));
1889 kibnal_data.kib_init = IBNAL_INIT_NOTHING;
1893 kibnal_api_startup (nal_t *nal, ptl_pid_t requested_pid,
1894 ptl_ni_limits_t *requested_limits,
1895 ptl_ni_limits_t *actual_limits)
1898 ptl_process_id_t process_id;
1899 int pkmem = atomic_read(&portal_kmemory);
1903 LASSERT (nal == &kibnal_api);
1905 if (nal->nal_refct != 0) {
1906 if (actual_limits != NULL)
1907 *actual_limits = kibnal_lib.libnal_ni.ni_actual_limits;
1908 /* This module got the first ref */
1913 LASSERT (kibnal_data.kib_init == IBNAL_INIT_NOTHING);
1915 memset (&kibnal_data, 0, sizeof (kibnal_data)); /* zero pointers, flags etc */
1917 do_gettimeofday(&tv);
1918 kibnal_data.kib_incarnation = (((__u64)tv.tv_sec) * 1000000) + tv.tv_usec;
1920 init_MUTEX (&kibnal_data.kib_nid_mutex);
1921 init_MUTEX_LOCKED (&kibnal_data.kib_listener_signal);
1923 rwlock_init(&kibnal_data.kib_global_lock);
1925 kibnal_data.kib_peer_hash_size = IBNAL_PEER_HASH_SIZE;
1926 PORTAL_ALLOC (kibnal_data.kib_peers,
1927 sizeof (struct list_head) * kibnal_data.kib_peer_hash_size);
1928 if (kibnal_data.kib_peers == NULL) {
1931 for (i = 0; i < kibnal_data.kib_peer_hash_size; i++)
1932 INIT_LIST_HEAD(&kibnal_data.kib_peers[i]);
1934 spin_lock_init (&kibnal_data.kib_reaper_lock);
1935 INIT_LIST_HEAD (&kibnal_data.kib_reaper_conns);
1936 init_waitqueue_head (&kibnal_data.kib_reaper_waitq);
1938 spin_lock_init (&kibnal_data.kib_connd_lock);
1939 INIT_LIST_HEAD (&kibnal_data.kib_connd_acceptq);
1940 INIT_LIST_HEAD (&kibnal_data.kib_connd_peers);
1941 init_waitqueue_head (&kibnal_data.kib_connd_waitq);
1943 spin_lock_init (&kibnal_data.kib_sched_lock);
1944 INIT_LIST_HEAD (&kibnal_data.kib_sched_txq);
1945 INIT_LIST_HEAD (&kibnal_data.kib_sched_rxq);
1946 init_waitqueue_head (&kibnal_data.kib_sched_waitq);
1948 spin_lock_init (&kibnal_data.kib_tx_lock);
1949 INIT_LIST_HEAD (&kibnal_data.kib_idle_txs);
1950 INIT_LIST_HEAD (&kibnal_data.kib_idle_nblk_txs);
1951 init_waitqueue_head(&kibnal_data.kib_idle_tx_waitq);
1953 PORTAL_ALLOC (kibnal_data.kib_tx_descs,
1954 IBNAL_TX_MSGS * sizeof(kib_tx_t));
1955 if (kibnal_data.kib_tx_descs == NULL) {
1956 CERROR ("Can't allocate tx descs\n");
1960 /* lists/ptrs/locks initialised */
1961 kibnal_data.kib_init = IBNAL_INIT_DATA;
1962 /*****************************************************/
1965 process_id.pid = requested_pid;
1966 process_id.nid = PTL_NID_ANY; /* don't know my NID yet */
1968 rc = lib_init(&kibnal_lib, nal, process_id,
1969 requested_limits, actual_limits);
1971 CERROR("lib_init failed: error %d\n", rc);
1975 /* lib interface initialised */
1976 kibnal_data.kib_init = IBNAL_INIT_LIB;
1977 /*****************************************************/
1979 for (i = 0; i < IBNAL_N_SCHED; i++) {
1980 rc = kibnal_thread_start (kibnal_scheduler,
1981 (void *)((unsigned long)i));
1983 CERROR("Can't spawn openibnal scheduler[%d]: %d\n",
1989 for (i = 0; i < IBNAL_N_CONND; i++) {
1990 rc = kibnal_thread_start (kibnal_connd,
1991 (void *)((unsigned long)i));
1993 CERROR("Can't spawn openibnal connd[%d]: %d\n",
1999 rc = kibnal_thread_start (kibnal_reaper, NULL);
2001 CERROR ("Can't spawn openibnal reaper: %d\n", rc);
2005 kibnal_data.kib_device = ib_device_get_by_index(0);
2006 if (kibnal_data.kib_device == NULL) {
2007 CERROR ("Can't open ib device 0\n");
2011 rc = ib_device_properties_get(kibnal_data.kib_device,
2012 &kibnal_data.kib_device_props);
2014 CERROR ("Can't get device props: %d\n", rc);
2018 CDEBUG(D_NET, "Max Initiator: %d Max Responder %d\n",
2019 kibnal_data.kib_device_props.max_initiator_per_qp,
2020 kibnal_data.kib_device_props.max_responder_per_qp);
2022 kibnal_data.kib_port = 0;
2023 for (i = 1; i <= 2; i++) {
2024 rc = ib_port_properties_get(kibnal_data.kib_device, i,
2025 &kibnal_data.kib_port_props);
2027 kibnal_data.kib_port = i;
2031 if (kibnal_data.kib_port == 0) {
2032 CERROR ("Can't find a port\n");
2036 rc = ib_pd_create(kibnal_data.kib_device,
2037 NULL, &kibnal_data.kib_pd);
2039 CERROR ("Can't create PD: %d\n", rc);
2043 /* flag PD initialised */
2044 kibnal_data.kib_init = IBNAL_INIT_PD;
2045 /*****************************************************/
2048 const int pool_size = IBNAL_NTX + IBNAL_NTX_NBLK;
2049 struct ib_fmr_pool_param params = {
2050 .max_pages_per_fmr = PTL_MTU/PAGE_SIZE,
2051 .access = (IB_ACCESS_LOCAL_WRITE |
2052 IB_ACCESS_REMOTE_WRITE |
2053 IB_ACCESS_REMOTE_READ),
2054 .pool_size = pool_size,
2055 .dirty_watermark = (pool_size * 3)/4,
2056 .flush_function = NULL,
2060 rc = ib_fmr_pool_create(kibnal_data.kib_pd, ¶ms,
2061 &kibnal_data.kib_fmr_pool);
2063 CERROR ("Can't create FMR pool size %d: %d\n",
2069 /* flag FMR pool initialised */
2070 kibnal_data.kib_init = IBNAL_INIT_FMR;
2072 /*****************************************************/
2074 rc = kibnal_setup_tx_descs();
2076 CERROR ("Can't register tx descs: %d\n", rc);
2080 /* flag TX descs initialised */
2081 kibnal_data.kib_init = IBNAL_INIT_TXD;
2082 /*****************************************************/
2085 struct ib_cq_callback callback = {
2086 .context = IBNAL_CALLBACK_CTXT,
2087 .policy = IB_CQ_PROVIDER_REARM,
2089 .entry = kibnal_callback,
2093 int nentries = IBNAL_CQ_ENTRIES;
2095 rc = ib_cq_create (kibnal_data.kib_device,
2096 &nentries, &callback, NULL,
2097 &kibnal_data.kib_cq);
2099 CERROR ("Can't create CQ: %d\n", rc);
2103 /* I only want solicited events */
2104 rc = ib_cq_request_notification(kibnal_data.kib_cq, 1);
2108 /* flag CQ initialised */
2109 kibnal_data.kib_init = IBNAL_INIT_CQ;
2110 /*****************************************************/
2112 rc = libcfs_nal_cmd_register(OPENIBNAL, &kibnal_cmd, NULL);
2114 CERROR ("Can't initialise command interface (rc = %d)\n", rc);
2118 /* flag everything initialised */
2119 kibnal_data.kib_init = IBNAL_INIT_ALL;
2120 /*****************************************************/
2122 printk(KERN_INFO "Lustre: OpenIB NAL loaded "
2123 "(initial mem %d)\n", pkmem);
2128 kibnal_api_shutdown (&kibnal_api);
2133 kibnal_module_fini (void)
2135 if (kibnal_tunables.kib_sysctl != NULL)
2136 unregister_sysctl_table (kibnal_tunables.kib_sysctl);
2137 PtlNIFini(kibnal_ni);
2139 ptl_unregister_nal(OPENIBNAL);
2143 kibnal_module_init (void)
2147 /* the following must be sizeof(int) for proc_dointvec() */
2148 LASSERT (sizeof(kibnal_tunables.kib_io_timeout) == sizeof(int));
2149 LASSERT (sizeof(kibnal_tunables.kib_listener_timeout) == sizeof(int));
2150 LASSERT (sizeof(kibnal_tunables.kib_backlog) == sizeof(int));
2151 LASSERT (sizeof(kibnal_tunables.kib_port) == sizeof(int));
2153 kibnal_api.nal_ni_init = kibnal_api_startup;
2154 kibnal_api.nal_ni_fini = kibnal_api_shutdown;
2156 /* Initialise dynamic tunables to defaults once only */
2157 kibnal_tunables.kib_io_timeout = IBNAL_IO_TIMEOUT;
2158 kibnal_tunables.kib_listener_timeout = IBNAL_LISTENER_TIMEOUT;
2159 kibnal_tunables.kib_backlog = IBNAL_BACKLOG;
2160 kibnal_tunables.kib_port = IBNAL_PORT;
2162 rc = ptl_register_nal(OPENIBNAL, &kibnal_api);
2164 CERROR("Can't register IBNAL: %d\n", rc);
2165 return (-ENOMEM); /* or something... */
2168 /* Pure gateways want the NAL started up at module load time... */
2169 rc = PtlNIInit(OPENIBNAL, LUSTRE_SRV_PTL_PID, NULL, NULL, &kibnal_ni);
2170 if (rc != PTL_OK && rc != PTL_IFACE_DUP) {
2171 ptl_unregister_nal(OPENIBNAL);
2175 kibnal_tunables.kib_sysctl =
2176 register_sysctl_table (kibnal_top_ctl_table, 0);
2177 if (kibnal_tunables.kib_sysctl == NULL) {
2178 CERROR("Can't register sysctl table\n");
2179 PtlNIFini(kibnal_ni);
2180 ptl_unregister_nal(OPENIBNAL);
2187 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
2188 MODULE_DESCRIPTION("Kernel OpenIB NAL v0.01");
2189 MODULE_LICENSE("GPL");
2191 module_init(kibnal_module_init);
2192 module_exit(kibnal_module_fini);