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 if (msg->ibm_srcnid == PTL_NID_ANY) {
165 CERROR("Bad src nid: "LPX64"\n", msg->ibm_srcnid);
169 switch (msg->ibm_type) {
171 CERROR("Unknown message type %x\n", msg->ibm_type);
174 case IBNAL_MSG_SVCQRY:
178 case IBNAL_MSG_SVCRSP:
179 if (msg_nob < hdr_size + sizeof(msg->ibm_u.svcrsp)) {
180 CERROR("Short SVCRSP: %d(%d)\n", msg_nob,
181 (int)(hdr_size + sizeof(msg->ibm_u.svcrsp)));
185 __swab64s(&msg->ibm_u.svcrsp.ibsr_svc_id);
186 __swab16s(&msg->ibm_u.svcrsp.ibsr_svc_pkey);
190 case IBNAL_MSG_CONNREQ:
191 case IBNAL_MSG_CONNACK:
192 if (msg_nob < hdr_size + sizeof(msg->ibm_u.connparams)) {
193 CERROR("Short CONNREQ: %d(%d)\n", msg_nob,
194 (int)(hdr_size + sizeof(msg->ibm_u.connparams)));
198 __swab32s(&msg->ibm_u.connparams.ibcp_queue_depth);
201 case IBNAL_MSG_IMMEDIATE:
202 if (msg_nob < offsetof(kib_msg_t, ibm_u.immediate.ibim_payload[0])) {
203 CERROR("Short IMMEDIATE: %d(%d)\n", msg_nob,
204 (int)offsetof(kib_msg_t, ibm_u.immediate.ibim_payload[0]));
209 case IBNAL_MSG_PUT_RDMA:
210 case IBNAL_MSG_GET_RDMA:
211 if (msg_nob < hdr_size + sizeof(msg->ibm_u.rdma)) {
212 CERROR("Short RDMA req: %d(%d)\n", msg_nob,
213 (int)(hdr_size + sizeof(msg->ibm_u.rdma)));
217 __swab32s(&msg->ibm_u.rdma.ibrm_desc.rd_key);
218 __swab32s(&msg->ibm_u.rdma.ibrm_desc.rd_nob);
219 __swab64s(&msg->ibm_u.rdma.ibrm_desc.rd_addr);
223 case IBNAL_MSG_PUT_DONE:
224 case IBNAL_MSG_GET_DONE:
225 if (msg_nob < hdr_size + sizeof(msg->ibm_u.completion)) {
226 CERROR("Short RDMA completion: %d(%d)\n", msg_nob,
227 (int)(hdr_size + sizeof(msg->ibm_u.completion)));
231 __swab32s(&msg->ibm_u.completion.ibcm_status);
238 kibnal_sock_write (struct socket *sock, void *buffer, int nob)
241 mm_segment_t oldmm = get_fs();
246 struct msghdr msg = {
253 .msg_flags = MSG_DONTWAIT
256 /* We've set up the socket's send buffer to be large enough for
257 * everything we send, so a single non-blocking send should
258 * complete without error. */
261 rc = sock_sendmsg(sock, &msg, iov.iov_len);
274 kibnal_sock_read (struct socket *sock, void *buffer, int nob, int timeout)
277 mm_segment_t oldmm = get_fs();
278 long ticks = timeout * HZ;
290 struct msghdr msg = {
300 /* Set receive timeout to remaining time */
301 tv = (struct timeval) {
302 .tv_sec = ticks / HZ,
303 .tv_usec = ((ticks % HZ) * 1000000) / HZ
306 rc = sock_setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO,
307 (char *)&tv, sizeof(tv));
310 CERROR("Can't set socket recv timeout %d: %d\n",
317 rc = sock_recvmsg(sock, &msg, iov.iov_len, 0);
318 ticks -= jiffies - then;
325 return -ECONNABORTED;
327 buffer = ((char *)buffer) + rc;
339 kibnal_create_sock(struct socket **sockp)
344 mm_segment_t oldmm = get_fs();
346 rc = sock_create(PF_INET, SOCK_STREAM, 0, &sock);
348 CERROR("Can't create socket: %d\n", rc);
352 /* Ensure sends will not block */
353 option = 2 * sizeof(kib_msg_t);
355 rc = sock_setsockopt(sock, SOL_SOCKET, SO_SNDBUF,
356 (char *)&option, sizeof(option));
359 CERROR("Can't set send buffer %d: %d\n", option, rc);
365 rc = sock_setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
366 (char *)&option, sizeof(option));
369 CERROR("Can't set SO_REUSEADDR: %d\n", rc);
382 kibnal_pause(int ticks)
384 set_current_state(TASK_UNINTERRUPTIBLE);
385 schedule_timeout(ticks);
389 kibnal_connect_sock(kib_peer_t *peer, struct socket **sockp)
391 struct sockaddr_in locaddr;
392 struct sockaddr_in srvaddr;
397 for (port = 1023; port >= 512; port--) {
399 memset(&locaddr, 0, sizeof(locaddr));
400 locaddr.sin_family = AF_INET;
401 locaddr.sin_port = htons(port);
402 locaddr.sin_addr.s_addr = htonl(INADDR_ANY);
404 memset (&srvaddr, 0, sizeof (srvaddr));
405 srvaddr.sin_family = AF_INET;
406 srvaddr.sin_port = htons (peer->ibp_port);
407 srvaddr.sin_addr.s_addr = htonl (peer->ibp_ip);
409 rc = kibnal_create_sock(&sock);
413 rc = sock->ops->bind(sock,
414 (struct sockaddr *)&locaddr, sizeof(locaddr));
418 if (rc == -EADDRINUSE) {
419 CDEBUG(D_NET, "Port %d already in use\n", port);
423 CERROR("Can't bind to reserved port %d: %d\n", port, rc);
427 rc = sock->ops->connect(sock,
428 (struct sockaddr *)&srvaddr, sizeof(srvaddr),
437 if (rc != -EADDRNOTAVAIL) {
438 CERROR("Can't connect port %d to %u.%u.%u.%u/%d: %d\n",
439 port, HIPQUAD(peer->ibp_ip), peer->ibp_port, rc);
443 CDEBUG(D_NET, "Port %d not available for %u.%u.%u.%u/%d\n",
444 port, HIPQUAD(peer->ibp_ip), peer->ibp_port);
448 return -EHOSTUNREACH;
452 kibnal_make_svcqry (kib_conn_t *conn)
454 kib_peer_t *peer = conn->ibc_peer;
460 LASSERT (conn->ibc_connreq != NULL);
461 msg = &conn->ibc_connreq->cr_msg;
463 kibnal_init_msg(msg, IBNAL_MSG_SVCQRY, 0);
464 kibnal_pack_msg(msg, 0, peer->ibp_nid, 0);
466 rc = kibnal_connect_sock(peer, &sock);
470 rc = kibnal_sock_write(sock, msg, msg->ibm_nob);
472 CERROR("Error %d sending svcqry to "
473 LPX64"@%u.%u.%u.%u/%d\n", rc,
474 peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
478 nob = offsetof(kib_msg_t, ibm_u) + sizeof(msg->ibm_u.svcrsp);
479 rc = kibnal_sock_read(sock, msg, nob, kibnal_tunables.kib_io_timeout);
481 CERROR("Error %d receiving svcrsp from "
482 LPX64"@%u.%u.%u.%u/%d\n", rc,
483 peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
487 rc = kibnal_unpack_msg(msg, nob);
489 CERROR("Error %d unpacking svcrsp from "
490 LPX64"@%u.%u.%u.%u/%d\n", rc,
491 peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
495 if (msg->ibm_type != IBNAL_MSG_SVCRSP) {
496 CERROR("Unexpected response type %d from "
497 LPX64"@%u.%u.%u.%u/%d\n", msg->ibm_type,
498 peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
503 if (msg->ibm_dstnid != kibnal_lib.libnal_ni.ni_pid.nid ||
504 msg->ibm_dststamp != kibnal_data.kib_incarnation) {
505 CERROR("Unexpected dst NID/stamp "LPX64"/"LPX64" from "
506 LPX64"@%u.%u.%u.%u/%d\n",
507 msg->ibm_dstnid, msg->ibm_dststamp,
508 peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
513 if (msg->ibm_srcnid != peer->ibp_nid) {
514 CERROR("Unexpected src NID "LPX64" from "
515 LPX64"@%u.%u.%u.%u/%d\n", msg->ibm_srcnid,
516 peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
521 conn->ibc_incarnation = msg->ibm_srcstamp;
522 conn->ibc_connreq->cr_svcrsp = msg->ibm_u.svcrsp;
529 kibnal_handle_svcqry (struct socket *sock)
531 struct sockaddr_in addr;
533 unsigned int peer_port;
541 rc = sock->ops->getname(sock, (struct sockaddr *)&addr, &len, 2);
543 CERROR("Can't get peer's IP: %d\n", rc);
547 peer_ip = ntohl(addr.sin_addr.s_addr);
548 peer_port = ntohs(addr.sin_port);
550 if (peer_port >= 1024) {
551 CERROR("Refusing unprivileged connection from %u.%u.%u.%u/%d\n",
552 HIPQUAD(peer_ip), peer_port);
556 PORTAL_ALLOC(msg, sizeof(*msg));
558 CERROR("Can't allocate msgs for %u.%u.%u.%u/%d\n",
559 HIPQUAD(peer_ip), peer_port);
563 rc = kibnal_sock_read(sock, msg, offsetof(kib_msg_t, ibm_u),
564 kibnal_tunables.kib_listener_timeout);
566 CERROR("Error %d receiving svcqry from %u.%u.%u.%u/%d\n",
567 rc, HIPQUAD(peer_ip), peer_port);
571 rc = kibnal_unpack_msg(msg, offsetof(kib_msg_t, ibm_u));
573 CERROR("Error %d unpacking svcqry from %u.%u.%u.%u/%d\n",
574 rc, HIPQUAD(peer_ip), peer_port);
578 if (msg->ibm_type != IBNAL_MSG_SVCQRY) {
579 CERROR("Unexpected message %d from %u.%u.%u.%u/%d\n",
580 msg->ibm_type, HIPQUAD(peer_ip), peer_port);
584 if (msg->ibm_dstnid != kibnal_lib.libnal_ni.ni_pid.nid) {
585 CERROR("Unexpected dstnid "LPX64"(expected "LPX64" "
586 "from %u.%u.%u.%u/%d\n", msg->ibm_dstnid,
587 kibnal_lib.libnal_ni.ni_pid.nid,
588 HIPQUAD(peer_ip), peer_port);
592 srcnid = msg->ibm_srcnid;
593 srcstamp = msg->ibm_srcstamp;
595 kibnal_init_msg(msg, IBNAL_MSG_SVCRSP, sizeof(msg->ibm_u.svcrsp));
597 msg->ibm_u.svcrsp.ibsr_svc_id = kibnal_data.kib_svc_id;
598 memcpy(msg->ibm_u.svcrsp.ibsr_svc_gid, kibnal_data.kib_svc_gid,
599 sizeof(kibnal_data.kib_svc_gid));
600 msg->ibm_u.svcrsp.ibsr_svc_pkey = kibnal_data.kib_svc_pkey;
602 kibnal_pack_msg(msg, 0, srcnid, srcstamp);
604 rc = kibnal_sock_write (sock, msg, msg->ibm_nob);
606 CERROR("Error %d replying to svcqry from %u.%u.%u.%u/%d\n",
607 rc, HIPQUAD(peer_ip), peer_port);
612 PORTAL_FREE(msg, sizeof(*msg));
616 kibnal_free_acceptsock (kib_acceptsock_t *as)
618 sock_release(as->ibas_sock);
619 PORTAL_FREE(as, sizeof(*as));
623 kibnal_ip_listener(void *arg)
625 struct sockaddr_in addr;
628 kib_acceptsock_t *as;
634 /* Parent thread holds kib_nid_mutex, and is, or is about to
635 * block on kib_listener_signal */
637 port = kibnal_tunables.kib_port;
638 snprintf(name, sizeof(name), "kibnal_lstn%03d", port);
639 kportal_daemonize(name);
640 kportal_blockallsigs();
642 init_waitqueue_entry(&wait, current);
644 rc = kibnal_create_sock(&sock);
648 memset(&addr, 0, sizeof(addr));
649 addr.sin_family = AF_INET;
650 addr.sin_port = htons(port);
651 addr.sin_addr.s_addr = INADDR_ANY;
653 rc = sock->ops->bind(sock, (struct sockaddr *)&addr, sizeof(addr));
655 CERROR("Can't bind to port %d\n", port);
659 rc = sock->ops->listen(sock, kibnal_tunables.kib_backlog);
661 CERROR("Can't set listen backlog %d: %d\n",
662 kibnal_tunables.kib_backlog, rc);
666 LASSERT (kibnal_data.kib_listener_sock == NULL);
667 kibnal_data.kib_listener_sock = sock;
669 /* unblock waiting parent */
670 LASSERT (kibnal_data.kib_listener_shutdown == 0);
671 up(&kibnal_data.kib_listener_signal);
673 /* Wake me any time something happens on my socket */
674 add_wait_queue(sock->sk->sk_sleep, &wait);
677 while (kibnal_data.kib_listener_shutdown == 0) {
680 PORTAL_ALLOC(as, sizeof(*as));
682 CERROR("Out of Memory: pausing...\n");
686 as->ibas_sock = NULL;
689 if (as->ibas_sock == NULL) {
690 as->ibas_sock = sock_alloc();
691 if (as->ibas_sock == NULL) {
692 CERROR("Can't allocate socket: pausing...\n");
696 /* XXX this should add a ref to sock->ops->owner, if
697 * TCP could be a module */
698 as->ibas_sock->type = sock->type;
699 as->ibas_sock->ops = sock->ops;
702 set_current_state(TASK_INTERRUPTIBLE);
704 rc = sock->ops->accept(sock, as->ibas_sock, O_NONBLOCK);
706 /* Sleep for socket activity? */
708 kibnal_data.kib_listener_shutdown == 0)
711 set_current_state(TASK_RUNNING);
714 spin_lock_irqsave(&kibnal_data.kib_connd_lock, flags);
716 list_add_tail(&as->ibas_list,
717 &kibnal_data.kib_connd_acceptq);
719 spin_unlock_irqrestore(&kibnal_data.kib_connd_lock, flags);
720 wake_up(&kibnal_data.kib_connd_waitq);
727 CERROR("Accept failed: %d, pausing...\n", rc);
733 if (as->ibas_sock != NULL)
734 sock_release(as->ibas_sock);
735 PORTAL_FREE(as, sizeof(*as));
739 remove_wait_queue(sock->sk->sk_sleep, &wait);
742 kibnal_data.kib_listener_sock = NULL;
744 /* set completion status and unblock thread waiting for me
745 * (parent on startup failure, executioner on normal shutdown) */
746 kibnal_data.kib_listener_shutdown = rc;
747 up(&kibnal_data.kib_listener_signal);
753 kibnal_start_ip_listener (void)
758 CDEBUG(D_NET, "Starting listener\n");
760 /* Called holding kib_nid_mutex: listener stopped */
761 LASSERT (kibnal_data.kib_listener_sock == NULL);
763 kibnal_data.kib_listener_shutdown = 0;
764 pid = kernel_thread(kibnal_ip_listener, NULL, 0);
766 CERROR("Can't spawn listener: %ld\n", pid);
770 /* Block until listener has started up. */
771 down(&kibnal_data.kib_listener_signal);
773 rc = kibnal_data.kib_listener_shutdown;
774 LASSERT ((rc != 0) == (kibnal_data.kib_listener_sock == NULL));
776 CDEBUG((rc == 0) ? D_WARNING : D_ERROR,
777 "Listener %s: pid:%ld port:%d backlog:%d\n",
778 (rc == 0) ? "started OK" : "startup failed",
779 pid, kibnal_tunables.kib_port, kibnal_tunables.kib_backlog);
785 kibnal_stop_ip_listener(int clear_acceptq)
787 struct list_head zombie_accepts;
788 kib_acceptsock_t *as;
791 CDEBUG(D_NET, "Stopping listener\n");
793 /* Called holding kib_nid_mutex: listener running */
794 LASSERT (kibnal_data.kib_listener_sock != NULL);
796 kibnal_data.kib_listener_shutdown = 1;
797 wake_up_all(kibnal_data.kib_listener_sock->sk->sk_sleep);
799 /* Block until listener has torn down. */
800 down(&kibnal_data.kib_listener_signal);
802 LASSERT (kibnal_data.kib_listener_sock == NULL);
803 CDEBUG(D_WARNING, "Listener stopped\n");
808 /* Close any unhandled accepts */
809 spin_lock_irqsave(&kibnal_data.kib_connd_lock, flags);
811 list_add(&zombie_accepts, &kibnal_data.kib_connd_acceptq);
812 list_del_init(&kibnal_data.kib_connd_acceptq);
814 spin_unlock_irqrestore(&kibnal_data.kib_connd_lock, flags);
816 while (!list_empty(&zombie_accepts)) {
817 as = list_entry(zombie_accepts.next,
818 kib_acceptsock_t, ibas_list);
819 list_del(&as->ibas_list);
820 kibnal_free_acceptsock(as);
825 kibnal_listener_procint(ctl_table *table, int write, struct file *filp,
826 void *buffer, size_t *lenp)
828 int *tunable = (int *)table->data;
832 /* No race with nal initialisation since the nal is setup all the time
833 * it's loaded. When that changes, change this! */
834 LASSERT (kibnal_data.kib_init == IBNAL_INIT_ALL);
836 down(&kibnal_data.kib_nid_mutex);
838 LASSERT (tunable == &kibnal_tunables.kib_port ||
839 tunable == &kibnal_tunables.kib_backlog);
842 rc = proc_dointvec(table, write, filp, buffer, lenp);
845 (*tunable != old_val ||
846 kibnal_data.kib_listener_sock == NULL)) {
848 if (kibnal_data.kib_listener_sock != NULL)
849 kibnal_stop_ip_listener(0);
851 rc = kibnal_start_ip_listener();
853 CERROR("Unable to restart listener with new tunable:"
854 " reverting to old value\n");
856 kibnal_start_ip_listener();
860 up(&kibnal_data.kib_nid_mutex);
862 LASSERT (kibnal_data.kib_init == IBNAL_INIT_ALL);
867 kibnal_start_ib_listener (void)
871 LASSERT (kibnal_data.kib_listen_handle == NULL);
873 kibnal_data.kib_svc_id = ib_cm_service_assign();
874 CDEBUG(D_NET, "svc id "LPX64"\n", kibnal_data.kib_svc_id);
876 rc = ib_cached_gid_get(kibnal_data.kib_device,
877 kibnal_data.kib_port, 0,
878 kibnal_data.kib_svc_gid);
880 CERROR("Can't get port %d GID: %d\n",
881 kibnal_data.kib_port, rc);
885 rc = ib_cached_pkey_get(kibnal_data.kib_device,
886 kibnal_data.kib_port, 0,
887 &kibnal_data.kib_svc_pkey);
889 CERROR ("Can't get port %d PKEY: %d\n",
890 kibnal_data.kib_port, rc);
894 rc = ib_cm_listen(kibnal_data.kib_svc_id,
895 TS_IB_CM_SERVICE_EXACT_MASK,
896 kibnal_passive_conn_callback, NULL,
897 &kibnal_data.kib_listen_handle);
899 kibnal_data.kib_listen_handle = NULL;
900 CERROR ("Can't create IB listener: %d\n", rc);
904 LASSERT (kibnal_data.kib_listen_handle != NULL);
909 kibnal_stop_ib_listener (void)
913 LASSERT (kibnal_data.kib_listen_handle != NULL);
915 rc = ib_cm_listen_stop (kibnal_data.kib_listen_handle);
917 CERROR("Error stopping IB listener: %d\n", rc);
919 kibnal_data.kib_listen_handle = NULL;
923 kibnal_set_mynid (ptl_nid_t nid)
925 lib_ni_t *ni = &kibnal_lib.libnal_ni;
928 CDEBUG(D_IOCTL, "setting mynid to "LPX64" (old nid="LPX64")\n",
929 nid, ni->ni_pid.nid);
931 down (&kibnal_data.kib_nid_mutex);
933 if (nid == kibnal_data.kib_nid) {
934 /* no change of NID */
935 up (&kibnal_data.kib_nid_mutex);
939 CDEBUG(D_NET, "NID "LPX64"("LPX64")\n",
940 kibnal_data.kib_nid, nid);
942 if (kibnal_data.kib_listener_sock != NULL)
943 kibnal_stop_ip_listener(1);
945 if (kibnal_data.kib_listen_handle != NULL)
946 kibnal_stop_ib_listener();
948 ni->ni_pid.nid = nid;
949 kibnal_data.kib_incarnation++;
951 /* Delete all existing peers and their connections after new
952 * NID/incarnation set to ensure no old connections in our brave new
954 kibnal_del_peer (PTL_NID_ANY, 0);
956 if (ni->ni_pid.nid != PTL_NID_ANY) {
957 /* got a new NID to install */
958 rc = kibnal_start_ib_listener();
960 CERROR("Can't start IB listener: %d\n", rc);
964 rc = kibnal_start_ip_listener();
966 CERROR("Can't start IP listener: %d\n", rc);
971 up(&kibnal_data.kib_nid_mutex);
975 kibnal_stop_ib_listener();
977 ni->ni_pid.nid = PTL_NID_ANY;
978 kibnal_data.kib_incarnation++;
980 kibnal_del_peer (PTL_NID_ANY, 0);
981 up(&kibnal_data.kib_nid_mutex);
986 kibnal_create_peer (ptl_nid_t nid)
990 LASSERT (nid != PTL_NID_ANY);
992 PORTAL_ALLOC (peer, sizeof (*peer));
996 memset(peer, 0, sizeof(*peer)); /* zero flags etc */
999 atomic_set (&peer->ibp_refcount, 1); /* 1 ref for caller */
1001 INIT_LIST_HEAD (&peer->ibp_list); /* not in the peer table yet */
1002 INIT_LIST_HEAD (&peer->ibp_conns);
1003 INIT_LIST_HEAD (&peer->ibp_tx_queue);
1005 peer->ibp_reconnect_time = jiffies;
1006 peer->ibp_reconnect_interval = IBNAL_MIN_RECONNECT_INTERVAL;
1008 atomic_inc (&kibnal_data.kib_npeers);
1009 CDEBUG(D_NET, "peer %p "LPX64"\n", peer, nid);
1015 kibnal_destroy_peer (kib_peer_t *peer)
1017 CDEBUG (D_NET, "peer "LPX64" %p deleted\n", peer->ibp_nid, peer);
1019 LASSERT (atomic_read (&peer->ibp_refcount) == 0);
1020 LASSERT (peer->ibp_persistence == 0);
1021 LASSERT (!kibnal_peer_active(peer));
1022 LASSERT (peer->ibp_connecting == 0);
1023 LASSERT (list_empty (&peer->ibp_conns));
1024 LASSERT (list_empty (&peer->ibp_tx_queue));
1026 PORTAL_FREE (peer, sizeof (*peer));
1028 /* NB a peer's connections keep a reference on their peer until
1029 * they are destroyed, so we can be assured that _all_ state to do
1030 * with this peer has been cleaned up when its refcount drops to
1032 atomic_dec (&kibnal_data.kib_npeers);
1036 kibnal_put_peer (kib_peer_t *peer)
1038 CDEBUG (D_OTHER, "putting peer[%p] -> "LPX64" (%d)\n",
1039 peer, peer->ibp_nid,
1040 atomic_read (&peer->ibp_refcount));
1042 LASSERT (atomic_read (&peer->ibp_refcount) > 0);
1043 if (!atomic_dec_and_test (&peer->ibp_refcount))
1046 kibnal_destroy_peer (peer);
1050 kibnal_find_peer_locked (ptl_nid_t nid)
1052 struct list_head *peer_list = kibnal_nid2peerlist (nid);
1053 struct list_head *tmp;
1056 list_for_each (tmp, peer_list) {
1058 peer = list_entry (tmp, kib_peer_t, ibp_list);
1060 LASSERT (peer->ibp_persistence != 0 || /* persistent peer */
1061 peer->ibp_connecting != 0 || /* creating conns */
1062 !list_empty (&peer->ibp_conns)); /* active conn */
1064 if (peer->ibp_nid != nid)
1067 CDEBUG(D_NET, "got peer [%p] -> "LPX64" (%d)\n",
1068 peer, nid, atomic_read (&peer->ibp_refcount));
1075 kibnal_get_peer (ptl_nid_t nid)
1079 read_lock (&kibnal_data.kib_global_lock);
1080 peer = kibnal_find_peer_locked (nid);
1081 if (peer != NULL) /* +1 ref for caller? */
1082 atomic_inc (&peer->ibp_refcount);
1083 read_unlock (&kibnal_data.kib_global_lock);
1089 kibnal_unlink_peer_locked (kib_peer_t *peer)
1091 LASSERT (peer->ibp_persistence == 0);
1092 LASSERT (list_empty(&peer->ibp_conns));
1094 LASSERT (kibnal_peer_active(peer));
1095 list_del_init (&peer->ibp_list);
1096 /* lose peerlist's ref */
1097 kibnal_put_peer (peer);
1101 kibnal_get_peer_info (int index, ptl_nid_t *nidp, __u32 *ipp, int *portp,
1105 struct list_head *ptmp;
1108 read_lock (&kibnal_data.kib_global_lock);
1110 for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
1112 list_for_each (ptmp, &kibnal_data.kib_peers[i]) {
1114 peer = list_entry (ptmp, kib_peer_t, ibp_list);
1115 LASSERT (peer->ibp_persistence != 0 ||
1116 peer->ibp_connecting != 0 ||
1117 !list_empty (&peer->ibp_conns));
1122 *nidp = peer->ibp_nid;
1123 *ipp = peer->ibp_ip;
1124 *portp = peer->ibp_port;
1125 *persistencep = peer->ibp_persistence;
1127 read_unlock (&kibnal_data.kib_global_lock);
1132 read_unlock (&kibnal_data.kib_global_lock);
1137 kibnal_add_persistent_peer (ptl_nid_t nid, __u32 ip, int port)
1139 unsigned long flags;
1143 if (nid == PTL_NID_ANY)
1146 peer = kibnal_create_peer (nid);
1150 write_lock_irqsave (&kibnal_data.kib_global_lock, flags);
1152 peer2 = kibnal_find_peer_locked (nid);
1153 if (peer2 != NULL) {
1154 kibnal_put_peer (peer);
1157 /* peer table takes existing ref on peer */
1158 list_add_tail (&peer->ibp_list,
1159 kibnal_nid2peerlist (nid));
1163 peer->ibp_port = port;
1164 peer->ibp_persistence++;
1166 write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
1171 kibnal_del_peer_locked (kib_peer_t *peer, int single_share)
1173 struct list_head *ctmp;
1174 struct list_head *cnxt;
1178 peer->ibp_persistence = 0;
1179 else if (peer->ibp_persistence > 0)
1180 peer->ibp_persistence--;
1182 if (peer->ibp_persistence != 0)
1185 if (list_empty(&peer->ibp_conns)) {
1186 kibnal_unlink_peer_locked(peer);
1188 list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
1189 conn = list_entry(ctmp, kib_conn_t, ibc_list);
1191 kibnal_close_conn_locked (conn, 0);
1193 /* NB peer is no longer persistent; closing its last conn
1196 /* NB peer now unlinked; might even be freed if the peer table had the
1197 * last ref on it. */
1201 kibnal_del_peer (ptl_nid_t nid, int single_share)
1203 unsigned long flags;
1204 struct list_head *ptmp;
1205 struct list_head *pnxt;
1212 write_lock_irqsave (&kibnal_data.kib_global_lock, flags);
1214 if (nid != PTL_NID_ANY)
1215 lo = hi = kibnal_nid2peerlist(nid) - kibnal_data.kib_peers;
1218 hi = kibnal_data.kib_peer_hash_size - 1;
1221 for (i = lo; i <= hi; i++) {
1222 list_for_each_safe (ptmp, pnxt, &kibnal_data.kib_peers[i]) {
1223 peer = list_entry (ptmp, kib_peer_t, ibp_list);
1224 LASSERT (peer->ibp_persistence != 0 ||
1225 peer->ibp_connecting != 0 ||
1226 !list_empty (&peer->ibp_conns));
1228 if (!(nid == PTL_NID_ANY || peer->ibp_nid == nid))
1231 kibnal_del_peer_locked (peer, single_share);
1232 rc = 0; /* matched something */
1239 write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
1245 kibnal_get_conn_by_idx (int index)
1248 struct list_head *ptmp;
1250 struct list_head *ctmp;
1253 read_lock (&kibnal_data.kib_global_lock);
1255 for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
1256 list_for_each (ptmp, &kibnal_data.kib_peers[i]) {
1258 peer = list_entry (ptmp, kib_peer_t, ibp_list);
1259 LASSERT (peer->ibp_persistence > 0 ||
1260 peer->ibp_connecting != 0 ||
1261 !list_empty (&peer->ibp_conns));
1263 list_for_each (ctmp, &peer->ibp_conns) {
1267 conn = list_entry (ctmp, kib_conn_t, ibc_list);
1268 CDEBUG(D_NET, "++conn[%p] state %d -> "LPX64" (%d)\n",
1269 conn, conn->ibc_state, conn->ibc_peer->ibp_nid,
1270 atomic_read (&conn->ibc_refcount));
1271 atomic_inc (&conn->ibc_refcount);
1272 read_unlock (&kibnal_data.kib_global_lock);
1278 read_unlock (&kibnal_data.kib_global_lock);
1283 kibnal_create_conn (void)
1293 struct ib_qp_create_param qp_create;
1294 struct ib_qp_attribute qp_attr;
1297 PORTAL_ALLOC (conn, sizeof (*conn));
1299 CERROR ("Can't allocate connection\n");
1303 /* zero flags, NULL pointers etc... */
1304 memset (conn, 0, sizeof (*conn));
1306 INIT_LIST_HEAD (&conn->ibc_tx_queue);
1307 INIT_LIST_HEAD (&conn->ibc_active_txs);
1308 spin_lock_init (&conn->ibc_lock);
1310 atomic_inc (&kibnal_data.kib_nconns);
1311 /* well not really, but I call destroy() on failure, which decrements */
1313 PORTAL_ALLOC (conn->ibc_rxs, IBNAL_RX_MSGS * sizeof (kib_rx_t));
1314 if (conn->ibc_rxs == NULL)
1316 memset (conn->ibc_rxs, 0, IBNAL_RX_MSGS * sizeof(kib_rx_t));
1318 rc = kibnal_alloc_pages(&conn->ibc_rx_pages,
1320 IB_ACCESS_LOCAL_WRITE);
1324 vaddr_base = vaddr = conn->ibc_rx_pages->ibp_vaddr;
1326 for (i = ipage = page_offset = 0; i < IBNAL_RX_MSGS; i++) {
1327 struct page *page = conn->ibc_rx_pages->ibp_pages[ipage];
1328 kib_rx_t *rx = &conn->ibc_rxs[i];
1331 rx->rx_vaddr = vaddr;
1332 rx->rx_msg = (kib_msg_t *)(((char *)page_address(page)) + page_offset);
1334 vaddr += IBNAL_MSG_SIZE;
1335 LASSERT (vaddr <= vaddr_base + IBNAL_RX_MSG_BYTES);
1337 page_offset += IBNAL_MSG_SIZE;
1338 LASSERT (page_offset <= PAGE_SIZE);
1340 if (page_offset == PAGE_SIZE) {
1343 LASSERT (ipage <= IBNAL_RX_MSG_PAGES);
1347 params.qp_create = (struct ib_qp_create_param) {
1349 /* Sends have an optional RDMA */
1350 .max_outstanding_send_request = 2 * IBNAL_MSG_QUEUE_SIZE,
1351 .max_outstanding_receive_request = IBNAL_MSG_QUEUE_SIZE,
1352 .max_send_gather_element = 1,
1353 .max_receive_scatter_element = 1,
1355 .pd = kibnal_data.kib_pd,
1356 .send_queue = kibnal_data.kib_cq,
1357 .receive_queue = kibnal_data.kib_cq,
1358 .send_policy = IB_WQ_SIGNAL_SELECTABLE,
1359 .receive_policy = IB_WQ_SIGNAL_SELECTABLE,
1361 .transport = IB_TRANSPORT_RC,
1362 .device_specific = NULL,
1365 rc = ib_qp_create (¶ms.qp_create, &conn->ibc_qp, &conn->ibc_qpn);
1367 CERROR ("Failed to create queue pair: %d\n", rc);
1371 /* Mark QP created */
1372 conn->ibc_state = IBNAL_CONN_INIT_QP;
1374 params.qp_attr = (struct ib_qp_attribute) {
1375 .state = IB_QP_STATE_INIT,
1376 .port = kibnal_data.kib_port,
1377 .enable_rdma_read = 1,
1378 .enable_rdma_write = 1,
1379 .valid_fields = (IB_QP_ATTRIBUTE_STATE |
1380 IB_QP_ATTRIBUTE_PORT |
1381 IB_QP_ATTRIBUTE_PKEY_INDEX |
1382 IB_QP_ATTRIBUTE_RDMA_ATOMIC_ENABLE),
1384 rc = ib_qp_modify(conn->ibc_qp, ¶ms.qp_attr);
1386 CERROR ("Failed to modify queue pair: %d\n", rc);
1390 /* 1 ref for caller */
1391 atomic_set (&conn->ibc_refcount, 1);
1395 kibnal_destroy_conn (conn);
1400 kibnal_destroy_conn (kib_conn_t *conn)
1404 CDEBUG (D_NET, "connection %p\n", conn);
1406 LASSERT (atomic_read (&conn->ibc_refcount) == 0);
1407 LASSERT (list_empty(&conn->ibc_tx_queue));
1408 LASSERT (list_empty(&conn->ibc_active_txs));
1409 LASSERT (conn->ibc_nsends_posted == 0);
1410 LASSERT (conn->ibc_connreq == NULL);
1412 switch (conn->ibc_state) {
1413 case IBNAL_CONN_ZOMBIE:
1414 /* called after connection sequence initiated */
1416 case IBNAL_CONN_INIT_QP:
1417 rc = ib_qp_destroy(conn->ibc_qp);
1419 CERROR("Can't destroy QP: %d\n", rc);
1422 case IBNAL_CONN_INIT_NOTHING:
1429 if (conn->ibc_rx_pages != NULL)
1430 kibnal_free_pages(conn->ibc_rx_pages);
1432 if (conn->ibc_rxs != NULL)
1433 PORTAL_FREE(conn->ibc_rxs,
1434 IBNAL_RX_MSGS * sizeof(kib_rx_t));
1436 if (conn->ibc_peer != NULL)
1437 kibnal_put_peer(conn->ibc_peer);
1439 PORTAL_FREE(conn, sizeof (*conn));
1441 atomic_dec(&kibnal_data.kib_nconns);
1443 if (atomic_read (&kibnal_data.kib_nconns) == 0 &&
1444 kibnal_data.kib_shutdown) {
1445 /* I just nuked the last connection on shutdown; wake up
1446 * everyone so they can exit. */
1447 wake_up_all(&kibnal_data.kib_sched_waitq);
1448 wake_up_all(&kibnal_data.kib_reaper_waitq);
1453 kibnal_put_conn (kib_conn_t *conn)
1455 unsigned long flags;
1457 CDEBUG (D_NET, "putting conn[%p] state %d -> "LPX64" (%d)\n",
1458 conn, conn->ibc_state, conn->ibc_peer->ibp_nid,
1459 atomic_read (&conn->ibc_refcount));
1461 LASSERT (atomic_read (&conn->ibc_refcount) > 0);
1462 if (!atomic_dec_and_test (&conn->ibc_refcount))
1465 /* last ref only goes on zombies */
1466 LASSERT (conn->ibc_state == IBNAL_CONN_ZOMBIE);
1468 spin_lock_irqsave (&kibnal_data.kib_reaper_lock, flags);
1470 list_add (&conn->ibc_list, &kibnal_data.kib_reaper_conns);
1471 wake_up (&kibnal_data.kib_reaper_waitq);
1473 spin_unlock_irqrestore (&kibnal_data.kib_reaper_lock, flags);
1477 kibnal_close_peer_conns_locked (kib_peer_t *peer, int why)
1480 struct list_head *ctmp;
1481 struct list_head *cnxt;
1484 list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
1485 conn = list_entry (ctmp, kib_conn_t, ibc_list);
1488 kibnal_close_conn_locked (conn, why);
1495 kibnal_close_stale_conns_locked (kib_peer_t *peer, __u64 incarnation)
1498 struct list_head *ctmp;
1499 struct list_head *cnxt;
1502 list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
1503 conn = list_entry (ctmp, kib_conn_t, ibc_list);
1505 if (conn->ibc_incarnation == incarnation)
1508 CDEBUG(D_NET, "Closing stale conn nid:"LPX64" incarnation:"LPX64"("LPX64")\n",
1509 peer->ibp_nid, conn->ibc_incarnation, incarnation);
1512 kibnal_close_conn_locked (conn, -ESTALE);
1519 kibnal_close_matching_conns (ptl_nid_t nid)
1521 unsigned long flags;
1523 struct list_head *ptmp;
1524 struct list_head *pnxt;
1530 write_lock_irqsave (&kibnal_data.kib_global_lock, flags);
1532 if (nid != PTL_NID_ANY)
1533 lo = hi = kibnal_nid2peerlist(nid) - kibnal_data.kib_peers;
1536 hi = kibnal_data.kib_peer_hash_size - 1;
1539 for (i = lo; i <= hi; i++) {
1540 list_for_each_safe (ptmp, pnxt, &kibnal_data.kib_peers[i]) {
1542 peer = list_entry (ptmp, kib_peer_t, ibp_list);
1543 LASSERT (peer->ibp_persistence != 0 ||
1544 peer->ibp_connecting != 0 ||
1545 !list_empty (&peer->ibp_conns));
1547 if (!(nid == PTL_NID_ANY || nid == peer->ibp_nid))
1550 count += kibnal_close_peer_conns_locked (peer, 0);
1554 write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
1556 /* wildcards always succeed */
1557 if (nid == PTL_NID_ANY)
1560 return (count == 0 ? -ENOENT : 0);
1564 kibnal_cmd(struct portals_cfg *pcfg, void * private)
1568 LASSERT (pcfg != NULL);
1570 switch(pcfg->pcfg_command) {
1571 case NAL_CMD_GET_PEER: {
1575 int share_count = 0;
1577 rc = kibnal_get_peer_info(pcfg->pcfg_count,
1578 &nid, &ip, &port, &share_count);
1579 pcfg->pcfg_nid = nid;
1580 pcfg->pcfg_size = 0;
1582 pcfg->pcfg_misc = port;
1583 pcfg->pcfg_count = 0;
1584 pcfg->pcfg_wait = share_count;
1587 case NAL_CMD_ADD_PEER: {
1588 rc = kibnal_add_persistent_peer (pcfg->pcfg_nid,
1589 pcfg->pcfg_id, /* IP */
1590 pcfg->pcfg_misc); /* port */
1593 case NAL_CMD_DEL_PEER: {
1594 rc = kibnal_del_peer (pcfg->pcfg_nid,
1595 /* flags == single_share */
1596 pcfg->pcfg_flags != 0);
1599 case NAL_CMD_GET_CONN: {
1600 kib_conn_t *conn = kibnal_get_conn_by_idx (pcfg->pcfg_count);
1606 pcfg->pcfg_nid = conn->ibc_peer->ibp_nid;
1608 pcfg->pcfg_misc = 0;
1609 pcfg->pcfg_flags = 0;
1610 kibnal_put_conn (conn);
1614 case NAL_CMD_CLOSE_CONNECTION: {
1615 rc = kibnal_close_matching_conns (pcfg->pcfg_nid);
1618 case NAL_CMD_REGISTER_MYNID: {
1619 if (pcfg->pcfg_nid == PTL_NID_ANY)
1622 rc = kibnal_set_mynid (pcfg->pcfg_nid);
1631 kibnal_free_pages (kib_pages_t *p)
1633 int npages = p->ibp_npages;
1637 if (p->ibp_mapped) {
1638 rc = ib_memory_deregister(p->ibp_handle);
1640 CERROR ("Deregister error: %d\n", rc);
1643 for (i = 0; i < npages; i++)
1644 if (p->ibp_pages[i] != NULL)
1645 __free_page(p->ibp_pages[i]);
1647 PORTAL_FREE (p, offsetof(kib_pages_t, ibp_pages[npages]));
1651 kibnal_alloc_pages (kib_pages_t **pp, int npages, int access)
1654 struct ib_physical_buffer *phys_pages;
1658 PORTAL_ALLOC(p, offsetof(kib_pages_t, ibp_pages[npages]));
1660 CERROR ("Can't allocate buffer %d\n", npages);
1664 memset (p, 0, offsetof(kib_pages_t, ibp_pages[npages]));
1665 p->ibp_npages = npages;
1667 for (i = 0; i < npages; i++) {
1668 p->ibp_pages[i] = alloc_page (GFP_KERNEL);
1669 if (p->ibp_pages[i] == NULL) {
1670 CERROR ("Can't allocate page %d of %d\n", i, npages);
1671 kibnal_free_pages(p);
1676 PORTAL_ALLOC(phys_pages, npages * sizeof(*phys_pages));
1677 if (phys_pages == NULL) {
1678 CERROR ("Can't allocate physarray for %d pages\n", npages);
1679 kibnal_free_pages(p);
1683 for (i = 0; i < npages; i++) {
1684 phys_pages[i].size = PAGE_SIZE;
1685 phys_pages[i].address =
1686 kibnal_page2phys(p->ibp_pages[i]);
1690 rc = ib_memory_register_physical(kibnal_data.kib_pd,
1693 npages * PAGE_SIZE, 0,
1699 PORTAL_FREE(phys_pages, npages * sizeof(*phys_pages));
1702 CERROR ("Error %d mapping %d pages\n", rc, npages);
1703 kibnal_free_pages(p);
1713 kibnal_setup_tx_descs (void)
1716 int page_offset = 0;
1724 /* pre-mapped messages are not bigger than 1 page */
1725 LASSERT (IBNAL_MSG_SIZE <= PAGE_SIZE);
1727 /* No fancy arithmetic when we do the buffer calculations */
1728 LASSERT (PAGE_SIZE % IBNAL_MSG_SIZE == 0);
1730 rc = kibnal_alloc_pages(&kibnal_data.kib_tx_pages,
1732 0); /* local read access only */
1736 vaddr = vaddr_base = kibnal_data.kib_tx_pages->ibp_vaddr;
1738 for (i = 0; i < IBNAL_TX_MSGS; i++) {
1739 page = kibnal_data.kib_tx_pages->ibp_pages[ipage];
1740 tx = &kibnal_data.kib_tx_descs[i];
1742 memset (tx, 0, sizeof(*tx)); /* zero flags etc */
1744 tx->tx_msg = (kib_msg_t *)(((char *)page_address(page)) + page_offset);
1745 tx->tx_vaddr = vaddr;
1746 tx->tx_isnblk = (i >= IBNAL_NTX);
1747 tx->tx_mapped = KIB_TX_UNMAPPED;
1749 CDEBUG(D_NET, "Tx[%d] %p->%p - "LPX64"\n",
1750 i, tx, tx->tx_msg, tx->tx_vaddr);
1753 list_add (&tx->tx_list,
1754 &kibnal_data.kib_idle_nblk_txs);
1756 list_add (&tx->tx_list,
1757 &kibnal_data.kib_idle_txs);
1759 vaddr += IBNAL_MSG_SIZE;
1760 LASSERT (vaddr <= vaddr_base + IBNAL_TX_MSG_BYTES);
1762 page_offset += IBNAL_MSG_SIZE;
1763 LASSERT (page_offset <= PAGE_SIZE);
1765 if (page_offset == PAGE_SIZE) {
1768 LASSERT (ipage <= IBNAL_TX_MSG_PAGES);
1776 kibnal_api_shutdown (nal_t *nal)
1781 if (nal->nal_refct != 0) {
1782 /* This module got the first ref */
1783 PORTAL_MODULE_UNUSE;
1787 CDEBUG(D_MALLOC, "before NAL cleanup: kmem %d\n",
1788 atomic_read (&portal_kmemory));
1790 LASSERT(nal == &kibnal_api);
1792 switch (kibnal_data.kib_init) {
1794 CERROR ("Unexpected state %d\n", kibnal_data.kib_init);
1797 case IBNAL_INIT_ALL:
1798 /* stop calls to nal_cmd */
1799 libcfs_nal_cmd_unregister(OPENIBNAL);
1802 /* resetting my NID unadvertises me, removes my
1803 * listener and nukes all current peers */
1804 kibnal_set_mynid (PTL_NID_ANY);
1806 /* Wait for all peer state to clean up */
1808 while (atomic_read (&kibnal_data.kib_npeers) != 0) {
1810 CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1811 "waiting for %d peers to close down\n",
1812 atomic_read (&kibnal_data.kib_npeers));
1813 set_current_state (TASK_INTERRUPTIBLE);
1814 schedule_timeout (HZ);
1819 rc = ib_cq_destroy (kibnal_data.kib_cq);
1821 CERROR ("Destroy CQ error: %d\n", rc);
1824 case IBNAL_INIT_TXD:
1825 kibnal_free_pages (kibnal_data.kib_tx_pages);
1828 case IBNAL_INIT_FMR:
1829 rc = ib_fmr_pool_destroy (kibnal_data.kib_fmr_pool);
1831 CERROR ("Destroy FMR pool error: %d\n", rc);
1835 rc = ib_pd_destroy(kibnal_data.kib_pd);
1837 CERROR ("Destroy PD error: %d\n", rc);
1840 case IBNAL_INIT_LIB:
1841 lib_fini(&kibnal_lib);
1844 case IBNAL_INIT_DATA:
1845 /* Module refcount only gets to zero when all peers
1846 * have been closed so all lists must be empty */
1847 LASSERT (atomic_read (&kibnal_data.kib_npeers) == 0);
1848 LASSERT (kibnal_data.kib_peers != NULL);
1849 for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
1850 LASSERT (list_empty (&kibnal_data.kib_peers[i]));
1852 LASSERT (atomic_read (&kibnal_data.kib_nconns) == 0);
1853 LASSERT (list_empty (&kibnal_data.kib_sched_rxq));
1854 LASSERT (list_empty (&kibnal_data.kib_sched_txq));
1855 LASSERT (list_empty (&kibnal_data.kib_reaper_conns));
1856 LASSERT (list_empty (&kibnal_data.kib_connd_peers));
1857 LASSERT (list_empty (&kibnal_data.kib_connd_acceptq));
1859 /* flag threads to terminate; wake and wait for them to die */
1860 kibnal_data.kib_shutdown = 1;
1861 wake_up_all (&kibnal_data.kib_sched_waitq);
1862 wake_up_all (&kibnal_data.kib_reaper_waitq);
1863 wake_up_all (&kibnal_data.kib_connd_waitq);
1866 while (atomic_read (&kibnal_data.kib_nthreads) != 0) {
1868 CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1869 "Waiting for %d threads to terminate\n",
1870 atomic_read (&kibnal_data.kib_nthreads));
1871 set_current_state (TASK_INTERRUPTIBLE);
1872 schedule_timeout (HZ);
1876 case IBNAL_INIT_NOTHING:
1880 if (kibnal_data.kib_tx_descs != NULL)
1881 PORTAL_FREE (kibnal_data.kib_tx_descs,
1882 IBNAL_TX_MSGS * sizeof(kib_tx_t));
1884 if (kibnal_data.kib_peers != NULL)
1885 PORTAL_FREE (kibnal_data.kib_peers,
1886 sizeof (struct list_head) *
1887 kibnal_data.kib_peer_hash_size);
1889 CDEBUG(D_MALLOC, "after NAL cleanup: kmem %d\n",
1890 atomic_read (&portal_kmemory));
1891 printk(KERN_INFO "Lustre: OpenIB NAL unloaded (final mem %d)\n",
1892 atomic_read(&portal_kmemory));
1894 kibnal_data.kib_init = IBNAL_INIT_NOTHING;
1898 kibnal_api_startup (nal_t *nal, ptl_pid_t requested_pid,
1899 ptl_ni_limits_t *requested_limits,
1900 ptl_ni_limits_t *actual_limits)
1903 ptl_process_id_t process_id;
1904 int pkmem = atomic_read(&portal_kmemory);
1908 LASSERT (nal == &kibnal_api);
1910 if (nal->nal_refct != 0) {
1911 if (actual_limits != NULL)
1912 *actual_limits = kibnal_lib.libnal_ni.ni_actual_limits;
1913 /* This module got the first ref */
1918 LASSERT (kibnal_data.kib_init == IBNAL_INIT_NOTHING);
1920 memset (&kibnal_data, 0, sizeof (kibnal_data)); /* zero pointers, flags etc */
1922 do_gettimeofday(&tv);
1923 kibnal_data.kib_incarnation = (((__u64)tv.tv_sec) * 1000000) + tv.tv_usec;
1925 init_MUTEX (&kibnal_data.kib_nid_mutex);
1926 init_MUTEX_LOCKED (&kibnal_data.kib_listener_signal);
1928 rwlock_init(&kibnal_data.kib_global_lock);
1930 kibnal_data.kib_peer_hash_size = IBNAL_PEER_HASH_SIZE;
1931 PORTAL_ALLOC (kibnal_data.kib_peers,
1932 sizeof (struct list_head) * kibnal_data.kib_peer_hash_size);
1933 if (kibnal_data.kib_peers == NULL) {
1936 for (i = 0; i < kibnal_data.kib_peer_hash_size; i++)
1937 INIT_LIST_HEAD(&kibnal_data.kib_peers[i]);
1939 spin_lock_init (&kibnal_data.kib_reaper_lock);
1940 INIT_LIST_HEAD (&kibnal_data.kib_reaper_conns);
1941 init_waitqueue_head (&kibnal_data.kib_reaper_waitq);
1943 spin_lock_init (&kibnal_data.kib_connd_lock);
1944 INIT_LIST_HEAD (&kibnal_data.kib_connd_acceptq);
1945 INIT_LIST_HEAD (&kibnal_data.kib_connd_peers);
1946 init_waitqueue_head (&kibnal_data.kib_connd_waitq);
1948 spin_lock_init (&kibnal_data.kib_sched_lock);
1949 INIT_LIST_HEAD (&kibnal_data.kib_sched_txq);
1950 INIT_LIST_HEAD (&kibnal_data.kib_sched_rxq);
1951 init_waitqueue_head (&kibnal_data.kib_sched_waitq);
1953 spin_lock_init (&kibnal_data.kib_tx_lock);
1954 INIT_LIST_HEAD (&kibnal_data.kib_idle_txs);
1955 INIT_LIST_HEAD (&kibnal_data.kib_idle_nblk_txs);
1956 init_waitqueue_head(&kibnal_data.kib_idle_tx_waitq);
1958 PORTAL_ALLOC (kibnal_data.kib_tx_descs,
1959 IBNAL_TX_MSGS * sizeof(kib_tx_t));
1960 if (kibnal_data.kib_tx_descs == NULL) {
1961 CERROR ("Can't allocate tx descs\n");
1965 /* lists/ptrs/locks initialised */
1966 kibnal_data.kib_init = IBNAL_INIT_DATA;
1967 /*****************************************************/
1970 process_id.pid = requested_pid;
1971 process_id.nid = PTL_NID_ANY; /* don't know my NID yet */
1973 rc = lib_init(&kibnal_lib, nal, process_id,
1974 requested_limits, actual_limits);
1976 CERROR("lib_init failed: error %d\n", rc);
1980 /* lib interface initialised */
1981 kibnal_data.kib_init = IBNAL_INIT_LIB;
1982 /*****************************************************/
1984 for (i = 0; i < IBNAL_N_SCHED; i++) {
1985 rc = kibnal_thread_start (kibnal_scheduler,
1986 (void *)((unsigned long)i));
1988 CERROR("Can't spawn openibnal scheduler[%d]: %d\n",
1994 for (i = 0; i < IBNAL_N_CONND; i++) {
1995 rc = kibnal_thread_start (kibnal_connd,
1996 (void *)((unsigned long)i));
1998 CERROR("Can't spawn openibnal connd[%d]: %d\n",
2004 rc = kibnal_thread_start (kibnal_reaper, NULL);
2006 CERROR ("Can't spawn openibnal reaper: %d\n", rc);
2010 kibnal_data.kib_device = ib_device_get_by_index(0);
2011 if (kibnal_data.kib_device == NULL) {
2012 CERROR ("Can't open ib device 0\n");
2016 rc = ib_device_properties_get(kibnal_data.kib_device,
2017 &kibnal_data.kib_device_props);
2019 CERROR ("Can't get device props: %d\n", rc);
2023 CDEBUG(D_NET, "Max Initiator: %d Max Responder %d\n",
2024 kibnal_data.kib_device_props.max_initiator_per_qp,
2025 kibnal_data.kib_device_props.max_responder_per_qp);
2027 kibnal_data.kib_port = 0;
2028 for (i = 1; i <= 2; i++) {
2029 rc = ib_port_properties_get(kibnal_data.kib_device, i,
2030 &kibnal_data.kib_port_props);
2032 kibnal_data.kib_port = i;
2036 if (kibnal_data.kib_port == 0) {
2037 CERROR ("Can't find a port\n");
2041 rc = ib_pd_create(kibnal_data.kib_device,
2042 NULL, &kibnal_data.kib_pd);
2044 CERROR ("Can't create PD: %d\n", rc);
2048 /* flag PD initialised */
2049 kibnal_data.kib_init = IBNAL_INIT_PD;
2050 /*****************************************************/
2053 const int pool_size = IBNAL_NTX + IBNAL_NTX_NBLK;
2054 struct ib_fmr_pool_param params = {
2055 .max_pages_per_fmr = PTL_MTU/PAGE_SIZE,
2056 .access = (IB_ACCESS_LOCAL_WRITE |
2057 IB_ACCESS_REMOTE_WRITE |
2058 IB_ACCESS_REMOTE_READ),
2059 .pool_size = pool_size,
2060 .dirty_watermark = (pool_size * 3)/4,
2061 .flush_function = NULL,
2065 rc = ib_fmr_pool_create(kibnal_data.kib_pd, ¶ms,
2066 &kibnal_data.kib_fmr_pool);
2068 CERROR ("Can't create FMR pool size %d: %d\n",
2074 /* flag FMR pool initialised */
2075 kibnal_data.kib_init = IBNAL_INIT_FMR;
2077 /*****************************************************/
2079 rc = kibnal_setup_tx_descs();
2081 CERROR ("Can't register tx descs: %d\n", rc);
2085 /* flag TX descs initialised */
2086 kibnal_data.kib_init = IBNAL_INIT_TXD;
2087 /*****************************************************/
2090 struct ib_cq_callback callback = {
2091 .context = IBNAL_CALLBACK_CTXT,
2092 .policy = IB_CQ_PROVIDER_REARM,
2094 .entry = kibnal_callback,
2098 int nentries = IBNAL_CQ_ENTRIES;
2100 rc = ib_cq_create (kibnal_data.kib_device,
2101 &nentries, &callback, NULL,
2102 &kibnal_data.kib_cq);
2104 CERROR ("Can't create CQ: %d\n", rc);
2108 /* I only want solicited events */
2109 rc = ib_cq_request_notification(kibnal_data.kib_cq, 1);
2113 /* flag CQ initialised */
2114 kibnal_data.kib_init = IBNAL_INIT_CQ;
2115 /*****************************************************/
2117 rc = libcfs_nal_cmd_register(OPENIBNAL, &kibnal_cmd, NULL);
2119 CERROR ("Can't initialise command interface (rc = %d)\n", rc);
2123 /* flag everything initialised */
2124 kibnal_data.kib_init = IBNAL_INIT_ALL;
2125 /*****************************************************/
2127 printk(KERN_INFO "Lustre: OpenIB NAL loaded "
2128 "(initial mem %d)\n", pkmem);
2133 kibnal_api_shutdown (&kibnal_api);
2138 kibnal_module_fini (void)
2140 if (kibnal_tunables.kib_sysctl != NULL)
2141 unregister_sysctl_table (kibnal_tunables.kib_sysctl);
2142 PtlNIFini(kibnal_ni);
2144 ptl_unregister_nal(OPENIBNAL);
2148 kibnal_module_init (void)
2152 /* the following must be sizeof(int) for proc_dointvec() */
2153 LASSERT (sizeof(kibnal_tunables.kib_io_timeout) == sizeof(int));
2154 LASSERT (sizeof(kibnal_tunables.kib_listener_timeout) == sizeof(int));
2155 LASSERT (sizeof(kibnal_tunables.kib_backlog) == sizeof(int));
2156 LASSERT (sizeof(kibnal_tunables.kib_port) == sizeof(int));
2158 kibnal_api.nal_ni_init = kibnal_api_startup;
2159 kibnal_api.nal_ni_fini = kibnal_api_shutdown;
2161 /* Initialise dynamic tunables to defaults once only */
2162 kibnal_tunables.kib_io_timeout = IBNAL_IO_TIMEOUT;
2163 kibnal_tunables.kib_listener_timeout = IBNAL_LISTENER_TIMEOUT;
2164 kibnal_tunables.kib_backlog = IBNAL_BACKLOG;
2165 kibnal_tunables.kib_port = IBNAL_PORT;
2167 rc = ptl_register_nal(OPENIBNAL, &kibnal_api);
2169 CERROR("Can't register IBNAL: %d\n", rc);
2170 return (-ENOMEM); /* or something... */
2173 /* Pure gateways want the NAL started up at module load time... */
2174 rc = PtlNIInit(OPENIBNAL, LUSTRE_SRV_PTL_PID, NULL, NULL, &kibnal_ni);
2175 if (rc != PTL_OK && rc != PTL_IFACE_DUP) {
2176 ptl_unregister_nal(OPENIBNAL);
2180 kibnal_tunables.kib_sysctl =
2181 register_sysctl_table (kibnal_top_ctl_table, 0);
2182 if (kibnal_tunables.kib_sysctl == NULL) {
2183 CERROR("Can't register sysctl table\n");
2184 PtlNIFini(kibnal_ni);
2185 ptl_unregister_nal(OPENIBNAL);
2192 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
2193 MODULE_DESCRIPTION("Kernel OpenIB NAL v0.01");
2194 MODULE_LICENSE("GPL");
2196 module_init(kibnal_module_init);
2197 module_exit(kibnal_module_fini);