Merge tag 'icc-6.7-rc5' of git://git.kernel.org/pub/scm/linux/kernel/git/djakov/icc...
[linux-2.6-microblaze.git] / drivers / infiniband / ulp / rtrs / rtrs-srv.c
1 // SPDX-License-Identifier: GPL-2.0-or-later
2 /*
3  * RDMA Transport Layer
4  *
5  * Copyright (c) 2014 - 2018 ProfitBricks GmbH. All rights reserved.
6  * Copyright (c) 2018 - 2019 1&1 IONOS Cloud GmbH. All rights reserved.
7  * Copyright (c) 2019 - 2020 1&1 IONOS SE. All rights reserved.
8  */
9
10 #undef pr_fmt
11 #define pr_fmt(fmt) KBUILD_MODNAME " L" __stringify(__LINE__) ": " fmt
12
13 #include <linux/module.h>
14
15 #include "rtrs-srv.h"
16 #include "rtrs-log.h"
17 #include <rdma/ib_cm.h>
18 #include <rdma/ib_verbs.h>
19 #include "rtrs-srv-trace.h"
20
21 MODULE_DESCRIPTION("RDMA Transport Server");
22 MODULE_LICENSE("GPL");
23
24 /* Must be power of 2, see mask from mr->page_size in ib_sg_to_pages() */
25 #define DEFAULT_MAX_CHUNK_SIZE (128 << 10)
26 #define DEFAULT_SESS_QUEUE_DEPTH 512
27 #define MAX_HDR_SIZE PAGE_SIZE
28
29 static struct rtrs_rdma_dev_pd dev_pd;
30 const struct class rtrs_dev_class = {
31         .name = "rtrs-server",
32 };
33 static struct rtrs_srv_ib_ctx ib_ctx;
34
35 static int __read_mostly max_chunk_size = DEFAULT_MAX_CHUNK_SIZE;
36 static int __read_mostly sess_queue_depth = DEFAULT_SESS_QUEUE_DEPTH;
37
38 static bool always_invalidate = true;
39 module_param(always_invalidate, bool, 0444);
40 MODULE_PARM_DESC(always_invalidate,
41                  "Invalidate memory registration for contiguous memory regions before accessing.");
42
43 module_param_named(max_chunk_size, max_chunk_size, int, 0444);
44 MODULE_PARM_DESC(max_chunk_size,
45                  "Max size for each IO request, when change the unit is in byte (default: "
46                  __stringify(DEFAULT_MAX_CHUNK_SIZE) "KB)");
47
48 module_param_named(sess_queue_depth, sess_queue_depth, int, 0444);
49 MODULE_PARM_DESC(sess_queue_depth,
50                  "Number of buffers for pending I/O requests to allocate per session. Maximum: "
51                  __stringify(MAX_SESS_QUEUE_DEPTH) " (default: "
52                  __stringify(DEFAULT_SESS_QUEUE_DEPTH) ")");
53
54 static cpumask_t cq_affinity_mask = { CPU_BITS_ALL };
55
56 static struct workqueue_struct *rtrs_wq;
57
58 static inline struct rtrs_srv_con *to_srv_con(struct rtrs_con *c)
59 {
60         return container_of(c, struct rtrs_srv_con, c);
61 }
62
63 static bool rtrs_srv_change_state(struct rtrs_srv_path *srv_path,
64                                   enum rtrs_srv_state new_state)
65 {
66         enum rtrs_srv_state old_state;
67         bool changed = false;
68         unsigned long flags;
69
70         spin_lock_irqsave(&srv_path->state_lock, flags);
71         old_state = srv_path->state;
72         switch (new_state) {
73         case RTRS_SRV_CONNECTED:
74                 if (old_state == RTRS_SRV_CONNECTING)
75                         changed = true;
76                 break;
77         case RTRS_SRV_CLOSING:
78                 if (old_state == RTRS_SRV_CONNECTING ||
79                     old_state == RTRS_SRV_CONNECTED)
80                         changed = true;
81                 break;
82         case RTRS_SRV_CLOSED:
83                 if (old_state == RTRS_SRV_CLOSING)
84                         changed = true;
85                 break;
86         default:
87                 break;
88         }
89         if (changed)
90                 srv_path->state = new_state;
91         spin_unlock_irqrestore(&srv_path->state_lock, flags);
92
93         return changed;
94 }
95
96 static void free_id(struct rtrs_srv_op *id)
97 {
98         if (!id)
99                 return;
100         kfree(id);
101 }
102
103 static void rtrs_srv_free_ops_ids(struct rtrs_srv_path *srv_path)
104 {
105         struct rtrs_srv_sess *srv = srv_path->srv;
106         int i;
107
108         if (srv_path->ops_ids) {
109                 for (i = 0; i < srv->queue_depth; i++)
110                         free_id(srv_path->ops_ids[i]);
111                 kfree(srv_path->ops_ids);
112                 srv_path->ops_ids = NULL;
113         }
114 }
115
116 static void rtrs_srv_rdma_done(struct ib_cq *cq, struct ib_wc *wc);
117
118 static struct ib_cqe io_comp_cqe = {
119         .done = rtrs_srv_rdma_done
120 };
121
122 static inline void rtrs_srv_inflight_ref_release(struct percpu_ref *ref)
123 {
124         struct rtrs_srv_path *srv_path = container_of(ref,
125                                                       struct rtrs_srv_path,
126                                                       ids_inflight_ref);
127
128         percpu_ref_exit(&srv_path->ids_inflight_ref);
129         complete(&srv_path->complete_done);
130 }
131
132 static int rtrs_srv_alloc_ops_ids(struct rtrs_srv_path *srv_path)
133 {
134         struct rtrs_srv_sess *srv = srv_path->srv;
135         struct rtrs_srv_op *id;
136         int i, ret;
137
138         srv_path->ops_ids = kcalloc(srv->queue_depth,
139                                     sizeof(*srv_path->ops_ids),
140                                     GFP_KERNEL);
141         if (!srv_path->ops_ids)
142                 goto err;
143
144         for (i = 0; i < srv->queue_depth; ++i) {
145                 id = kzalloc(sizeof(*id), GFP_KERNEL);
146                 if (!id)
147                         goto err;
148
149                 srv_path->ops_ids[i] = id;
150         }
151
152         ret = percpu_ref_init(&srv_path->ids_inflight_ref,
153                               rtrs_srv_inflight_ref_release, 0, GFP_KERNEL);
154         if (ret) {
155                 pr_err("Percpu reference init failed\n");
156                 goto err;
157         }
158         init_completion(&srv_path->complete_done);
159
160         return 0;
161
162 err:
163         rtrs_srv_free_ops_ids(srv_path);
164         return -ENOMEM;
165 }
166
167 static inline void rtrs_srv_get_ops_ids(struct rtrs_srv_path *srv_path)
168 {
169         percpu_ref_get(&srv_path->ids_inflight_ref);
170 }
171
172 static inline void rtrs_srv_put_ops_ids(struct rtrs_srv_path *srv_path)
173 {
174         percpu_ref_put(&srv_path->ids_inflight_ref);
175 }
176
177 static void rtrs_srv_reg_mr_done(struct ib_cq *cq, struct ib_wc *wc)
178 {
179         struct rtrs_srv_con *con = to_srv_con(wc->qp->qp_context);
180         struct rtrs_path *s = con->c.path;
181         struct rtrs_srv_path *srv_path = to_srv_path(s);
182
183         if (wc->status != IB_WC_SUCCESS) {
184                 rtrs_err(s, "REG MR failed: %s\n",
185                           ib_wc_status_msg(wc->status));
186                 close_path(srv_path);
187                 return;
188         }
189 }
190
191 static struct ib_cqe local_reg_cqe = {
192         .done = rtrs_srv_reg_mr_done
193 };
194
195 static int rdma_write_sg(struct rtrs_srv_op *id)
196 {
197         struct rtrs_path *s = id->con->c.path;
198         struct rtrs_srv_path *srv_path = to_srv_path(s);
199         dma_addr_t dma_addr = srv_path->dma_addr[id->msg_id];
200         struct rtrs_srv_mr *srv_mr;
201         struct ib_send_wr inv_wr;
202         struct ib_rdma_wr imm_wr;
203         struct ib_rdma_wr *wr = NULL;
204         enum ib_send_flags flags;
205         size_t sg_cnt;
206         int err, offset;
207         bool need_inval;
208         u32 rkey = 0;
209         struct ib_reg_wr rwr;
210         struct ib_sge *plist;
211         struct ib_sge list;
212
213         sg_cnt = le16_to_cpu(id->rd_msg->sg_cnt);
214         need_inval = le16_to_cpu(id->rd_msg->flags) & RTRS_MSG_NEED_INVAL_F;
215         if (sg_cnt != 1)
216                 return -EINVAL;
217
218         offset = 0;
219
220         wr              = &id->tx_wr;
221         plist           = &id->tx_sg;
222         plist->addr     = dma_addr + offset;
223         plist->length   = le32_to_cpu(id->rd_msg->desc[0].len);
224
225         /* WR will fail with length error
226          * if this is 0
227          */
228         if (plist->length == 0) {
229                 rtrs_err(s, "Invalid RDMA-Write sg list length 0\n");
230                 return -EINVAL;
231         }
232
233         plist->lkey = srv_path->s.dev->ib_pd->local_dma_lkey;
234         offset += plist->length;
235
236         wr->wr.sg_list  = plist;
237         wr->wr.num_sge  = 1;
238         wr->remote_addr = le64_to_cpu(id->rd_msg->desc[0].addr);
239         wr->rkey        = le32_to_cpu(id->rd_msg->desc[0].key);
240         if (rkey == 0)
241                 rkey = wr->rkey;
242         else
243                 /* Only one key is actually used */
244                 WARN_ON_ONCE(rkey != wr->rkey);
245
246         wr->wr.opcode = IB_WR_RDMA_WRITE;
247         wr->wr.wr_cqe   = &io_comp_cqe;
248         wr->wr.ex.imm_data = 0;
249         wr->wr.send_flags  = 0;
250
251         if (need_inval && always_invalidate) {
252                 wr->wr.next = &rwr.wr;
253                 rwr.wr.next = &inv_wr;
254                 inv_wr.next = &imm_wr.wr;
255         } else if (always_invalidate) {
256                 wr->wr.next = &rwr.wr;
257                 rwr.wr.next = &imm_wr.wr;
258         } else if (need_inval) {
259                 wr->wr.next = &inv_wr;
260                 inv_wr.next = &imm_wr.wr;
261         } else {
262                 wr->wr.next = &imm_wr.wr;
263         }
264         /*
265          * From time to time we have to post signaled sends,
266          * or send queue will fill up and only QP reset can help.
267          */
268         flags = (atomic_inc_return(&id->con->c.wr_cnt) % s->signal_interval) ?
269                 0 : IB_SEND_SIGNALED;
270
271         if (need_inval) {
272                 inv_wr.sg_list = NULL;
273                 inv_wr.num_sge = 0;
274                 inv_wr.opcode = IB_WR_SEND_WITH_INV;
275                 inv_wr.wr_cqe   = &io_comp_cqe;
276                 inv_wr.send_flags = 0;
277                 inv_wr.ex.invalidate_rkey = rkey;
278         }
279
280         imm_wr.wr.next = NULL;
281         if (always_invalidate) {
282                 struct rtrs_msg_rkey_rsp *msg;
283
284                 srv_mr = &srv_path->mrs[id->msg_id];
285                 rwr.wr.opcode = IB_WR_REG_MR;
286                 rwr.wr.wr_cqe = &local_reg_cqe;
287                 rwr.wr.num_sge = 0;
288                 rwr.mr = srv_mr->mr;
289                 rwr.wr.send_flags = 0;
290                 rwr.key = srv_mr->mr->rkey;
291                 rwr.access = (IB_ACCESS_LOCAL_WRITE |
292                               IB_ACCESS_REMOTE_WRITE);
293                 msg = srv_mr->iu->buf;
294                 msg->buf_id = cpu_to_le16(id->msg_id);
295                 msg->type = cpu_to_le16(RTRS_MSG_RKEY_RSP);
296                 msg->rkey = cpu_to_le32(srv_mr->mr->rkey);
297
298                 list.addr   = srv_mr->iu->dma_addr;
299                 list.length = sizeof(*msg);
300                 list.lkey   = srv_path->s.dev->ib_pd->local_dma_lkey;
301                 imm_wr.wr.sg_list = &list;
302                 imm_wr.wr.num_sge = 1;
303                 imm_wr.wr.opcode = IB_WR_SEND_WITH_IMM;
304                 ib_dma_sync_single_for_device(srv_path->s.dev->ib_dev,
305                                               srv_mr->iu->dma_addr,
306                                               srv_mr->iu->size, DMA_TO_DEVICE);
307         } else {
308                 imm_wr.wr.sg_list = NULL;
309                 imm_wr.wr.num_sge = 0;
310                 imm_wr.wr.opcode = IB_WR_RDMA_WRITE_WITH_IMM;
311         }
312         imm_wr.wr.send_flags = flags;
313         imm_wr.wr.ex.imm_data = cpu_to_be32(rtrs_to_io_rsp_imm(id->msg_id,
314                                                              0, need_inval));
315
316         imm_wr.wr.wr_cqe   = &io_comp_cqe;
317         ib_dma_sync_single_for_device(srv_path->s.dev->ib_dev, dma_addr,
318                                       offset, DMA_BIDIRECTIONAL);
319
320         err = ib_post_send(id->con->c.qp, &id->tx_wr.wr, NULL);
321         if (err)
322                 rtrs_err(s,
323                           "Posting RDMA-Write-Request to QP failed, err: %d\n",
324                           err);
325
326         return err;
327 }
328
329 /**
330  * send_io_resp_imm() - respond to client with empty IMM on failed READ/WRITE
331  *                      requests or on successful WRITE request.
332  * @con:        the connection to send back result
333  * @id:         the id associated with the IO
334  * @errno:      the error number of the IO.
335  *
336  * Return 0 on success, errno otherwise.
337  */
338 static int send_io_resp_imm(struct rtrs_srv_con *con, struct rtrs_srv_op *id,
339                             int errno)
340 {
341         struct rtrs_path *s = con->c.path;
342         struct rtrs_srv_path *srv_path = to_srv_path(s);
343         struct ib_send_wr inv_wr, *wr = NULL;
344         struct ib_rdma_wr imm_wr;
345         struct ib_reg_wr rwr;
346         struct rtrs_srv_mr *srv_mr;
347         bool need_inval = false;
348         enum ib_send_flags flags;
349         u32 imm;
350         int err;
351
352         if (id->dir == READ) {
353                 struct rtrs_msg_rdma_read *rd_msg = id->rd_msg;
354                 size_t sg_cnt;
355
356                 need_inval = le16_to_cpu(rd_msg->flags) &
357                                 RTRS_MSG_NEED_INVAL_F;
358                 sg_cnt = le16_to_cpu(rd_msg->sg_cnt);
359
360                 if (need_inval) {
361                         if (sg_cnt) {
362                                 inv_wr.wr_cqe   = &io_comp_cqe;
363                                 inv_wr.sg_list = NULL;
364                                 inv_wr.num_sge = 0;
365                                 inv_wr.opcode = IB_WR_SEND_WITH_INV;
366                                 inv_wr.send_flags = 0;
367                                 /* Only one key is actually used */
368                                 inv_wr.ex.invalidate_rkey =
369                                         le32_to_cpu(rd_msg->desc[0].key);
370                         } else {
371                                 WARN_ON_ONCE(1);
372                                 need_inval = false;
373                         }
374                 }
375         }
376
377         trace_send_io_resp_imm(id, need_inval, always_invalidate, errno);
378
379         if (need_inval && always_invalidate) {
380                 wr = &inv_wr;
381                 inv_wr.next = &rwr.wr;
382                 rwr.wr.next = &imm_wr.wr;
383         } else if (always_invalidate) {
384                 wr = &rwr.wr;
385                 rwr.wr.next = &imm_wr.wr;
386         } else if (need_inval) {
387                 wr = &inv_wr;
388                 inv_wr.next = &imm_wr.wr;
389         } else {
390                 wr = &imm_wr.wr;
391         }
392         /*
393          * From time to time we have to post signalled sends,
394          * or send queue will fill up and only QP reset can help.
395          */
396         flags = (atomic_inc_return(&con->c.wr_cnt) % s->signal_interval) ?
397                 0 : IB_SEND_SIGNALED;
398         imm = rtrs_to_io_rsp_imm(id->msg_id, errno, need_inval);
399         imm_wr.wr.next = NULL;
400         if (always_invalidate) {
401                 struct ib_sge list;
402                 struct rtrs_msg_rkey_rsp *msg;
403
404                 srv_mr = &srv_path->mrs[id->msg_id];
405                 rwr.wr.next = &imm_wr.wr;
406                 rwr.wr.opcode = IB_WR_REG_MR;
407                 rwr.wr.wr_cqe = &local_reg_cqe;
408                 rwr.wr.num_sge = 0;
409                 rwr.wr.send_flags = 0;
410                 rwr.mr = srv_mr->mr;
411                 rwr.key = srv_mr->mr->rkey;
412                 rwr.access = (IB_ACCESS_LOCAL_WRITE |
413                               IB_ACCESS_REMOTE_WRITE);
414                 msg = srv_mr->iu->buf;
415                 msg->buf_id = cpu_to_le16(id->msg_id);
416                 msg->type = cpu_to_le16(RTRS_MSG_RKEY_RSP);
417                 msg->rkey = cpu_to_le32(srv_mr->mr->rkey);
418
419                 list.addr   = srv_mr->iu->dma_addr;
420                 list.length = sizeof(*msg);
421                 list.lkey   = srv_path->s.dev->ib_pd->local_dma_lkey;
422                 imm_wr.wr.sg_list = &list;
423                 imm_wr.wr.num_sge = 1;
424                 imm_wr.wr.opcode = IB_WR_SEND_WITH_IMM;
425                 ib_dma_sync_single_for_device(srv_path->s.dev->ib_dev,
426                                               srv_mr->iu->dma_addr,
427                                               srv_mr->iu->size, DMA_TO_DEVICE);
428         } else {
429                 imm_wr.wr.sg_list = NULL;
430                 imm_wr.wr.num_sge = 0;
431                 imm_wr.wr.opcode = IB_WR_RDMA_WRITE_WITH_IMM;
432         }
433         imm_wr.wr.send_flags = flags;
434         imm_wr.wr.wr_cqe   = &io_comp_cqe;
435
436         imm_wr.wr.ex.imm_data = cpu_to_be32(imm);
437
438         err = ib_post_send(id->con->c.qp, wr, NULL);
439         if (err)
440                 rtrs_err_rl(s, "Posting RDMA-Reply to QP failed, err: %d\n",
441                              err);
442
443         return err;
444 }
445
446 void close_path(struct rtrs_srv_path *srv_path)
447 {
448         if (rtrs_srv_change_state(srv_path, RTRS_SRV_CLOSING))
449                 queue_work(rtrs_wq, &srv_path->close_work);
450         WARN_ON(srv_path->state != RTRS_SRV_CLOSING);
451 }
452
453 static inline const char *rtrs_srv_state_str(enum rtrs_srv_state state)
454 {
455         switch (state) {
456         case RTRS_SRV_CONNECTING:
457                 return "RTRS_SRV_CONNECTING";
458         case RTRS_SRV_CONNECTED:
459                 return "RTRS_SRV_CONNECTED";
460         case RTRS_SRV_CLOSING:
461                 return "RTRS_SRV_CLOSING";
462         case RTRS_SRV_CLOSED:
463                 return "RTRS_SRV_CLOSED";
464         default:
465                 return "UNKNOWN";
466         }
467 }
468
469 /**
470  * rtrs_srv_resp_rdma() - Finish an RDMA request
471  *
472  * @id:         Internal RTRS operation identifier
473  * @status:     Response Code sent to the other side for this operation.
474  *              0 = success, <=0 error
475  * Context: any
476  *
477  * Finish a RDMA operation. A message is sent to the client and the
478  * corresponding memory areas will be released.
479  */
480 bool rtrs_srv_resp_rdma(struct rtrs_srv_op *id, int status)
481 {
482         struct rtrs_srv_path *srv_path;
483         struct rtrs_srv_con *con;
484         struct rtrs_path *s;
485         int err;
486
487         if (WARN_ON(!id))
488                 return true;
489
490         con = id->con;
491         s = con->c.path;
492         srv_path = to_srv_path(s);
493
494         id->status = status;
495
496         if (srv_path->state != RTRS_SRV_CONNECTED) {
497                 rtrs_err_rl(s,
498                             "Sending I/O response failed,  server path %s is disconnected, path state %s\n",
499                             kobject_name(&srv_path->kobj),
500                             rtrs_srv_state_str(srv_path->state));
501                 goto out;
502         }
503         if (always_invalidate) {
504                 struct rtrs_srv_mr *mr = &srv_path->mrs[id->msg_id];
505
506                 ib_update_fast_reg_key(mr->mr, ib_inc_rkey(mr->mr->rkey));
507         }
508         if (atomic_sub_return(1, &con->c.sq_wr_avail) < 0) {
509                 rtrs_err(s, "IB send queue full: srv_path=%s cid=%d\n",
510                          kobject_name(&srv_path->kobj),
511                          con->c.cid);
512                 atomic_add(1, &con->c.sq_wr_avail);
513                 spin_lock(&con->rsp_wr_wait_lock);
514                 list_add_tail(&id->wait_list, &con->rsp_wr_wait_list);
515                 spin_unlock(&con->rsp_wr_wait_lock);
516                 return false;
517         }
518
519         if (status || id->dir == WRITE || !id->rd_msg->sg_cnt)
520                 err = send_io_resp_imm(con, id, status);
521         else
522                 err = rdma_write_sg(id);
523
524         if (err) {
525                 rtrs_err_rl(s, "IO response failed: %d: srv_path=%s\n", err,
526                             kobject_name(&srv_path->kobj));
527                 close_path(srv_path);
528         }
529 out:
530         rtrs_srv_put_ops_ids(srv_path);
531         return true;
532 }
533 EXPORT_SYMBOL(rtrs_srv_resp_rdma);
534
535 /**
536  * rtrs_srv_set_sess_priv() - Set private pointer in rtrs_srv.
537  * @srv:        Session pointer
538  * @priv:       The private pointer that is associated with the session.
539  */
540 void rtrs_srv_set_sess_priv(struct rtrs_srv_sess *srv, void *priv)
541 {
542         srv->priv = priv;
543 }
544 EXPORT_SYMBOL(rtrs_srv_set_sess_priv);
545
546 static void unmap_cont_bufs(struct rtrs_srv_path *srv_path)
547 {
548         int i;
549
550         for (i = 0; i < srv_path->mrs_num; i++) {
551                 struct rtrs_srv_mr *srv_mr;
552
553                 srv_mr = &srv_path->mrs[i];
554
555                 if (always_invalidate)
556                         rtrs_iu_free(srv_mr->iu, srv_path->s.dev->ib_dev, 1);
557
558                 ib_dereg_mr(srv_mr->mr);
559                 ib_dma_unmap_sg(srv_path->s.dev->ib_dev, srv_mr->sgt.sgl,
560                                 srv_mr->sgt.nents, DMA_BIDIRECTIONAL);
561                 sg_free_table(&srv_mr->sgt);
562         }
563         kfree(srv_path->mrs);
564 }
565
566 static int map_cont_bufs(struct rtrs_srv_path *srv_path)
567 {
568         struct rtrs_srv_sess *srv = srv_path->srv;
569         struct rtrs_path *ss = &srv_path->s;
570         int i, err, mrs_num;
571         unsigned int chunk_bits;
572         int chunks_per_mr = 1;
573         struct ib_mr *mr;
574         struct sg_table *sgt;
575
576         /*
577          * Here we map queue_depth chunks to MR.  Firstly we have to
578          * figure out how many chunks can we map per MR.
579          */
580         if (always_invalidate) {
581                 /*
582                  * in order to do invalidate for each chunks of memory, we needs
583                  * more memory regions.
584                  */
585                 mrs_num = srv->queue_depth;
586         } else {
587                 chunks_per_mr =
588                         srv_path->s.dev->ib_dev->attrs.max_fast_reg_page_list_len;
589                 mrs_num = DIV_ROUND_UP(srv->queue_depth, chunks_per_mr);
590                 chunks_per_mr = DIV_ROUND_UP(srv->queue_depth, mrs_num);
591         }
592
593         srv_path->mrs = kcalloc(mrs_num, sizeof(*srv_path->mrs), GFP_KERNEL);
594         if (!srv_path->mrs)
595                 return -ENOMEM;
596
597         for (srv_path->mrs_num = 0; srv_path->mrs_num < mrs_num;
598              srv_path->mrs_num++) {
599                 struct rtrs_srv_mr *srv_mr = &srv_path->mrs[srv_path->mrs_num];
600                 struct scatterlist *s;
601                 int nr, nr_sgt, chunks;
602
603                 sgt = &srv_mr->sgt;
604                 chunks = chunks_per_mr * srv_path->mrs_num;
605                 if (!always_invalidate)
606                         chunks_per_mr = min_t(int, chunks_per_mr,
607                                               srv->queue_depth - chunks);
608
609                 err = sg_alloc_table(sgt, chunks_per_mr, GFP_KERNEL);
610                 if (err)
611                         goto err;
612
613                 for_each_sg(sgt->sgl, s, chunks_per_mr, i)
614                         sg_set_page(s, srv->chunks[chunks + i],
615                                     max_chunk_size, 0);
616
617                 nr_sgt = ib_dma_map_sg(srv_path->s.dev->ib_dev, sgt->sgl,
618                                    sgt->nents, DMA_BIDIRECTIONAL);
619                 if (!nr_sgt) {
620                         err = -EINVAL;
621                         goto free_sg;
622                 }
623                 mr = ib_alloc_mr(srv_path->s.dev->ib_pd, IB_MR_TYPE_MEM_REG,
624                                  nr_sgt);
625                 if (IS_ERR(mr)) {
626                         err = PTR_ERR(mr);
627                         goto unmap_sg;
628                 }
629                 nr = ib_map_mr_sg(mr, sgt->sgl, nr_sgt,
630                                   NULL, max_chunk_size);
631                 if (nr != nr_sgt) {
632                         err = nr < 0 ? nr : -EINVAL;
633                         goto dereg_mr;
634                 }
635
636                 if (always_invalidate) {
637                         srv_mr->iu = rtrs_iu_alloc(1,
638                                         sizeof(struct rtrs_msg_rkey_rsp),
639                                         GFP_KERNEL, srv_path->s.dev->ib_dev,
640                                         DMA_TO_DEVICE, rtrs_srv_rdma_done);
641                         if (!srv_mr->iu) {
642                                 err = -ENOMEM;
643                                 rtrs_err(ss, "rtrs_iu_alloc(), err: %d\n", err);
644                                 goto dereg_mr;
645                         }
646                 }
647                 /* Eventually dma addr for each chunk can be cached */
648                 for_each_sg(sgt->sgl, s, nr_sgt, i)
649                         srv_path->dma_addr[chunks + i] = sg_dma_address(s);
650
651                 ib_update_fast_reg_key(mr, ib_inc_rkey(mr->rkey));
652                 srv_mr->mr = mr;
653         }
654
655         chunk_bits = ilog2(srv->queue_depth - 1) + 1;
656         srv_path->mem_bits = (MAX_IMM_PAYL_BITS - chunk_bits);
657
658         return 0;
659
660 dereg_mr:
661         ib_dereg_mr(mr);
662 unmap_sg:
663         ib_dma_unmap_sg(srv_path->s.dev->ib_dev, sgt->sgl,
664                         sgt->nents, DMA_BIDIRECTIONAL);
665 free_sg:
666         sg_free_table(sgt);
667 err:
668         unmap_cont_bufs(srv_path);
669
670         return err;
671 }
672
673 static void rtrs_srv_hb_err_handler(struct rtrs_con *c)
674 {
675         close_path(to_srv_path(c->path));
676 }
677
678 static void rtrs_srv_init_hb(struct rtrs_srv_path *srv_path)
679 {
680         rtrs_init_hb(&srv_path->s, &io_comp_cqe,
681                       RTRS_HB_INTERVAL_MS,
682                       RTRS_HB_MISSED_MAX,
683                       rtrs_srv_hb_err_handler,
684                       rtrs_wq);
685 }
686
687 static void rtrs_srv_start_hb(struct rtrs_srv_path *srv_path)
688 {
689         rtrs_start_hb(&srv_path->s);
690 }
691
692 static void rtrs_srv_stop_hb(struct rtrs_srv_path *srv_path)
693 {
694         rtrs_stop_hb(&srv_path->s);
695 }
696
697 static void rtrs_srv_info_rsp_done(struct ib_cq *cq, struct ib_wc *wc)
698 {
699         struct rtrs_srv_con *con = to_srv_con(wc->qp->qp_context);
700         struct rtrs_path *s = con->c.path;
701         struct rtrs_srv_path *srv_path = to_srv_path(s);
702         struct rtrs_iu *iu;
703
704         iu = container_of(wc->wr_cqe, struct rtrs_iu, cqe);
705         rtrs_iu_free(iu, srv_path->s.dev->ib_dev, 1);
706
707         if (wc->status != IB_WC_SUCCESS) {
708                 rtrs_err(s, "Sess info response send failed: %s\n",
709                           ib_wc_status_msg(wc->status));
710                 close_path(srv_path);
711                 return;
712         }
713         WARN_ON(wc->opcode != IB_WC_SEND);
714 }
715
716 static int rtrs_srv_path_up(struct rtrs_srv_path *srv_path)
717 {
718         struct rtrs_srv_sess *srv = srv_path->srv;
719         struct rtrs_srv_ctx *ctx = srv->ctx;
720         int up, ret = 0;
721
722         mutex_lock(&srv->paths_ev_mutex);
723         up = ++srv->paths_up;
724         if (up == 1)
725                 ret = ctx->ops.link_ev(srv, RTRS_SRV_LINK_EV_CONNECTED, NULL);
726         mutex_unlock(&srv->paths_ev_mutex);
727
728         /* Mark session as established */
729         if (!ret)
730                 srv_path->established = true;
731
732         return ret;
733 }
734
735 static void rtrs_srv_path_down(struct rtrs_srv_path *srv_path)
736 {
737         struct rtrs_srv_sess *srv = srv_path->srv;
738         struct rtrs_srv_ctx *ctx = srv->ctx;
739
740         if (!srv_path->established)
741                 return;
742
743         srv_path->established = false;
744         mutex_lock(&srv->paths_ev_mutex);
745         WARN_ON(!srv->paths_up);
746         if (--srv->paths_up == 0)
747                 ctx->ops.link_ev(srv, RTRS_SRV_LINK_EV_DISCONNECTED, srv->priv);
748         mutex_unlock(&srv->paths_ev_mutex);
749 }
750
751 static bool exist_pathname(struct rtrs_srv_ctx *ctx,
752                            const char *pathname, const uuid_t *path_uuid)
753 {
754         struct rtrs_srv_sess *srv;
755         struct rtrs_srv_path *srv_path;
756         bool found = false;
757
758         mutex_lock(&ctx->srv_mutex);
759         list_for_each_entry(srv, &ctx->srv_list, ctx_list) {
760                 mutex_lock(&srv->paths_mutex);
761
762                 /* when a client with same uuid and same sessname tried to add a path */
763                 if (uuid_equal(&srv->paths_uuid, path_uuid)) {
764                         mutex_unlock(&srv->paths_mutex);
765                         continue;
766                 }
767
768                 list_for_each_entry(srv_path, &srv->paths_list, s.entry) {
769                         if (strlen(srv_path->s.sessname) == strlen(pathname) &&
770                             !strcmp(srv_path->s.sessname, pathname)) {
771                                 found = true;
772                                 break;
773                         }
774                 }
775                 mutex_unlock(&srv->paths_mutex);
776                 if (found)
777                         break;
778         }
779         mutex_unlock(&ctx->srv_mutex);
780         return found;
781 }
782
783 static int post_recv_path(struct rtrs_srv_path *srv_path);
784 static int rtrs_rdma_do_reject(struct rdma_cm_id *cm_id, int errno);
785
786 static int process_info_req(struct rtrs_srv_con *con,
787                             struct rtrs_msg_info_req *msg)
788 {
789         struct rtrs_path *s = con->c.path;
790         struct rtrs_srv_path *srv_path = to_srv_path(s);
791         struct ib_send_wr *reg_wr = NULL;
792         struct rtrs_msg_info_rsp *rsp;
793         struct rtrs_iu *tx_iu;
794         struct ib_reg_wr *rwr;
795         int mri, err;
796         size_t tx_sz;
797
798         err = post_recv_path(srv_path);
799         if (err) {
800                 rtrs_err(s, "post_recv_path(), err: %d\n", err);
801                 return err;
802         }
803
804         if (strchr(msg->pathname, '/') || strchr(msg->pathname, '.')) {
805                 rtrs_err(s, "pathname cannot contain / and .\n");
806                 return -EINVAL;
807         }
808
809         if (exist_pathname(srv_path->srv->ctx,
810                            msg->pathname, &srv_path->srv->paths_uuid)) {
811                 rtrs_err(s, "pathname is duplicated: %s\n", msg->pathname);
812                 return -EPERM;
813         }
814         strscpy(srv_path->s.sessname, msg->pathname,
815                 sizeof(srv_path->s.sessname));
816
817         rwr = kcalloc(srv_path->mrs_num, sizeof(*rwr), GFP_KERNEL);
818         if (!rwr)
819                 return -ENOMEM;
820
821         tx_sz  = sizeof(*rsp);
822         tx_sz += sizeof(rsp->desc[0]) * srv_path->mrs_num;
823         tx_iu = rtrs_iu_alloc(1, tx_sz, GFP_KERNEL, srv_path->s.dev->ib_dev,
824                                DMA_TO_DEVICE, rtrs_srv_info_rsp_done);
825         if (!tx_iu) {
826                 err = -ENOMEM;
827                 goto rwr_free;
828         }
829
830         rsp = tx_iu->buf;
831         rsp->type = cpu_to_le16(RTRS_MSG_INFO_RSP);
832         rsp->sg_cnt = cpu_to_le16(srv_path->mrs_num);
833
834         for (mri = 0; mri < srv_path->mrs_num; mri++) {
835                 struct ib_mr *mr = srv_path->mrs[mri].mr;
836
837                 rsp->desc[mri].addr = cpu_to_le64(mr->iova);
838                 rsp->desc[mri].key  = cpu_to_le32(mr->rkey);
839                 rsp->desc[mri].len  = cpu_to_le32(mr->length);
840
841                 /*
842                  * Fill in reg MR request and chain them *backwards*
843                  */
844                 rwr[mri].wr.next = mri ? &rwr[mri - 1].wr : NULL;
845                 rwr[mri].wr.opcode = IB_WR_REG_MR;
846                 rwr[mri].wr.wr_cqe = &local_reg_cqe;
847                 rwr[mri].wr.num_sge = 0;
848                 rwr[mri].wr.send_flags = 0;
849                 rwr[mri].mr = mr;
850                 rwr[mri].key = mr->rkey;
851                 rwr[mri].access = (IB_ACCESS_LOCAL_WRITE |
852                                    IB_ACCESS_REMOTE_WRITE);
853                 reg_wr = &rwr[mri].wr;
854         }
855
856         err = rtrs_srv_create_path_files(srv_path);
857         if (err)
858                 goto iu_free;
859         kobject_get(&srv_path->kobj);
860         get_device(&srv_path->srv->dev);
861         err = rtrs_srv_change_state(srv_path, RTRS_SRV_CONNECTED);
862         if (!err) {
863                 rtrs_err(s, "rtrs_srv_change_state(), err: %d\n", err);
864                 goto iu_free;
865         }
866
867         rtrs_srv_start_hb(srv_path);
868
869         /*
870          * We do not account number of established connections at the current
871          * moment, we rely on the client, which should send info request when
872          * all connections are successfully established.  Thus, simply notify
873          * listener with a proper event if we are the first path.
874          */
875         err = rtrs_srv_path_up(srv_path);
876         if (err) {
877                 rtrs_err(s, "rtrs_srv_path_up(), err: %d\n", err);
878                 goto iu_free;
879         }
880
881         ib_dma_sync_single_for_device(srv_path->s.dev->ib_dev,
882                                       tx_iu->dma_addr,
883                                       tx_iu->size, DMA_TO_DEVICE);
884
885         /* Send info response */
886         err = rtrs_iu_post_send(&con->c, tx_iu, tx_sz, reg_wr);
887         if (err) {
888                 rtrs_err(s, "rtrs_iu_post_send(), err: %d\n", err);
889 iu_free:
890                 rtrs_iu_free(tx_iu, srv_path->s.dev->ib_dev, 1);
891         }
892 rwr_free:
893         kfree(rwr);
894
895         return err;
896 }
897
898 static void rtrs_srv_info_req_done(struct ib_cq *cq, struct ib_wc *wc)
899 {
900         struct rtrs_srv_con *con = to_srv_con(wc->qp->qp_context);
901         struct rtrs_path *s = con->c.path;
902         struct rtrs_srv_path *srv_path = to_srv_path(s);
903         struct rtrs_msg_info_req *msg;
904         struct rtrs_iu *iu;
905         int err;
906
907         WARN_ON(con->c.cid);
908
909         iu = container_of(wc->wr_cqe, struct rtrs_iu, cqe);
910         if (wc->status != IB_WC_SUCCESS) {
911                 rtrs_err(s, "Sess info request receive failed: %s\n",
912                           ib_wc_status_msg(wc->status));
913                 goto close;
914         }
915         WARN_ON(wc->opcode != IB_WC_RECV);
916
917         if (wc->byte_len < sizeof(*msg)) {
918                 rtrs_err(s, "Sess info request is malformed: size %d\n",
919                           wc->byte_len);
920                 goto close;
921         }
922         ib_dma_sync_single_for_cpu(srv_path->s.dev->ib_dev, iu->dma_addr,
923                                    iu->size, DMA_FROM_DEVICE);
924         msg = iu->buf;
925         if (le16_to_cpu(msg->type) != RTRS_MSG_INFO_REQ) {
926                 rtrs_err(s, "Sess info request is malformed: type %d\n",
927                           le16_to_cpu(msg->type));
928                 goto close;
929         }
930         err = process_info_req(con, msg);
931         if (err)
932                 goto close;
933
934 out:
935         rtrs_iu_free(iu, srv_path->s.dev->ib_dev, 1);
936         return;
937 close:
938         close_path(srv_path);
939         goto out;
940 }
941
942 static int post_recv_info_req(struct rtrs_srv_con *con)
943 {
944         struct rtrs_path *s = con->c.path;
945         struct rtrs_srv_path *srv_path = to_srv_path(s);
946         struct rtrs_iu *rx_iu;
947         int err;
948
949         rx_iu = rtrs_iu_alloc(1, sizeof(struct rtrs_msg_info_req),
950                                GFP_KERNEL, srv_path->s.dev->ib_dev,
951                                DMA_FROM_DEVICE, rtrs_srv_info_req_done);
952         if (!rx_iu)
953                 return -ENOMEM;
954         /* Prepare for getting info response */
955         err = rtrs_iu_post_recv(&con->c, rx_iu);
956         if (err) {
957                 rtrs_err(s, "rtrs_iu_post_recv(), err: %d\n", err);
958                 rtrs_iu_free(rx_iu, srv_path->s.dev->ib_dev, 1);
959                 return err;
960         }
961
962         return 0;
963 }
964
965 static int post_recv_io(struct rtrs_srv_con *con, size_t q_size)
966 {
967         int i, err;
968
969         for (i = 0; i < q_size; i++) {
970                 err = rtrs_post_recv_empty(&con->c, &io_comp_cqe);
971                 if (err)
972                         return err;
973         }
974
975         return 0;
976 }
977
978 static int post_recv_path(struct rtrs_srv_path *srv_path)
979 {
980         struct rtrs_srv_sess *srv = srv_path->srv;
981         struct rtrs_path *s = &srv_path->s;
982         size_t q_size;
983         int err, cid;
984
985         for (cid = 0; cid < srv_path->s.con_num; cid++) {
986                 if (cid == 0)
987                         q_size = SERVICE_CON_QUEUE_DEPTH;
988                 else
989                         q_size = srv->queue_depth;
990
991                 err = post_recv_io(to_srv_con(srv_path->s.con[cid]), q_size);
992                 if (err) {
993                         rtrs_err(s, "post_recv_io(), err: %d\n", err);
994                         return err;
995                 }
996         }
997
998         return 0;
999 }
1000
1001 static void process_read(struct rtrs_srv_con *con,
1002                          struct rtrs_msg_rdma_read *msg,
1003                          u32 buf_id, u32 off)
1004 {
1005         struct rtrs_path *s = con->c.path;
1006         struct rtrs_srv_path *srv_path = to_srv_path(s);
1007         struct rtrs_srv_sess *srv = srv_path->srv;
1008         struct rtrs_srv_ctx *ctx = srv->ctx;
1009         struct rtrs_srv_op *id;
1010
1011         size_t usr_len, data_len;
1012         void *data;
1013         int ret;
1014
1015         if (srv_path->state != RTRS_SRV_CONNECTED) {
1016                 rtrs_err_rl(s,
1017                              "Processing read request failed,  session is disconnected, sess state %s\n",
1018                              rtrs_srv_state_str(srv_path->state));
1019                 return;
1020         }
1021         if (msg->sg_cnt != 1 && msg->sg_cnt != 0) {
1022                 rtrs_err_rl(s,
1023                             "Processing read request failed, invalid message\n");
1024                 return;
1025         }
1026         rtrs_srv_get_ops_ids(srv_path);
1027         rtrs_srv_update_rdma_stats(srv_path->stats, off, READ);
1028         id = srv_path->ops_ids[buf_id];
1029         id->con         = con;
1030         id->dir         = READ;
1031         id->msg_id      = buf_id;
1032         id->rd_msg      = msg;
1033         usr_len = le16_to_cpu(msg->usr_len);
1034         data_len = off - usr_len;
1035         data = page_address(srv->chunks[buf_id]);
1036         ret = ctx->ops.rdma_ev(srv->priv, id, data, data_len,
1037                            data + data_len, usr_len);
1038
1039         if (ret) {
1040                 rtrs_err_rl(s,
1041                              "Processing read request failed, user module cb reported for msg_id %d, err: %d\n",
1042                              buf_id, ret);
1043                 goto send_err_msg;
1044         }
1045
1046         return;
1047
1048 send_err_msg:
1049         ret = send_io_resp_imm(con, id, ret);
1050         if (ret < 0) {
1051                 rtrs_err_rl(s,
1052                              "Sending err msg for failed RDMA-Write-Req failed, msg_id %d, err: %d\n",
1053                              buf_id, ret);
1054                 close_path(srv_path);
1055         }
1056         rtrs_srv_put_ops_ids(srv_path);
1057 }
1058
1059 static void process_write(struct rtrs_srv_con *con,
1060                           struct rtrs_msg_rdma_write *req,
1061                           u32 buf_id, u32 off)
1062 {
1063         struct rtrs_path *s = con->c.path;
1064         struct rtrs_srv_path *srv_path = to_srv_path(s);
1065         struct rtrs_srv_sess *srv = srv_path->srv;
1066         struct rtrs_srv_ctx *ctx = srv->ctx;
1067         struct rtrs_srv_op *id;
1068
1069         size_t data_len, usr_len;
1070         void *data;
1071         int ret;
1072
1073         if (srv_path->state != RTRS_SRV_CONNECTED) {
1074                 rtrs_err_rl(s,
1075                              "Processing write request failed,  session is disconnected, sess state %s\n",
1076                              rtrs_srv_state_str(srv_path->state));
1077                 return;
1078         }
1079         rtrs_srv_get_ops_ids(srv_path);
1080         rtrs_srv_update_rdma_stats(srv_path->stats, off, WRITE);
1081         id = srv_path->ops_ids[buf_id];
1082         id->con    = con;
1083         id->dir    = WRITE;
1084         id->msg_id = buf_id;
1085
1086         usr_len = le16_to_cpu(req->usr_len);
1087         data_len = off - usr_len;
1088         data = page_address(srv->chunks[buf_id]);
1089         ret = ctx->ops.rdma_ev(srv->priv, id, data, data_len,
1090                                data + data_len, usr_len);
1091         if (ret) {
1092                 rtrs_err_rl(s,
1093                              "Processing write request failed, user module callback reports err: %d\n",
1094                              ret);
1095                 goto send_err_msg;
1096         }
1097
1098         return;
1099
1100 send_err_msg:
1101         ret = send_io_resp_imm(con, id, ret);
1102         if (ret < 0) {
1103                 rtrs_err_rl(s,
1104                              "Processing write request failed, sending I/O response failed, msg_id %d, err: %d\n",
1105                              buf_id, ret);
1106                 close_path(srv_path);
1107         }
1108         rtrs_srv_put_ops_ids(srv_path);
1109 }
1110
1111 static void process_io_req(struct rtrs_srv_con *con, void *msg,
1112                            u32 id, u32 off)
1113 {
1114         struct rtrs_path *s = con->c.path;
1115         struct rtrs_srv_path *srv_path = to_srv_path(s);
1116         struct rtrs_msg_rdma_hdr *hdr;
1117         unsigned int type;
1118
1119         ib_dma_sync_single_for_cpu(srv_path->s.dev->ib_dev,
1120                                    srv_path->dma_addr[id],
1121                                    max_chunk_size, DMA_BIDIRECTIONAL);
1122         hdr = msg;
1123         type = le16_to_cpu(hdr->type);
1124
1125         switch (type) {
1126         case RTRS_MSG_WRITE:
1127                 process_write(con, msg, id, off);
1128                 break;
1129         case RTRS_MSG_READ:
1130                 process_read(con, msg, id, off);
1131                 break;
1132         default:
1133                 rtrs_err(s,
1134                           "Processing I/O request failed, unknown message type received: 0x%02x\n",
1135                           type);
1136                 goto err;
1137         }
1138
1139         return;
1140
1141 err:
1142         close_path(srv_path);
1143 }
1144
1145 static void rtrs_srv_inv_rkey_done(struct ib_cq *cq, struct ib_wc *wc)
1146 {
1147         struct rtrs_srv_mr *mr =
1148                 container_of(wc->wr_cqe, typeof(*mr), inv_cqe);
1149         struct rtrs_srv_con *con = to_srv_con(wc->qp->qp_context);
1150         struct rtrs_path *s = con->c.path;
1151         struct rtrs_srv_path *srv_path = to_srv_path(s);
1152         struct rtrs_srv_sess *srv = srv_path->srv;
1153         u32 msg_id, off;
1154         void *data;
1155
1156         if (wc->status != IB_WC_SUCCESS) {
1157                 rtrs_err(s, "Failed IB_WR_LOCAL_INV: %s\n",
1158                           ib_wc_status_msg(wc->status));
1159                 close_path(srv_path);
1160         }
1161         msg_id = mr->msg_id;
1162         off = mr->msg_off;
1163         data = page_address(srv->chunks[msg_id]) + off;
1164         process_io_req(con, data, msg_id, off);
1165 }
1166
1167 static int rtrs_srv_inv_rkey(struct rtrs_srv_con *con,
1168                               struct rtrs_srv_mr *mr)
1169 {
1170         struct ib_send_wr wr = {
1171                 .opcode             = IB_WR_LOCAL_INV,
1172                 .wr_cqe             = &mr->inv_cqe,
1173                 .send_flags         = IB_SEND_SIGNALED,
1174                 .ex.invalidate_rkey = mr->mr->rkey,
1175         };
1176         mr->inv_cqe.done = rtrs_srv_inv_rkey_done;
1177
1178         return ib_post_send(con->c.qp, &wr, NULL);
1179 }
1180
1181 static void rtrs_rdma_process_wr_wait_list(struct rtrs_srv_con *con)
1182 {
1183         spin_lock(&con->rsp_wr_wait_lock);
1184         while (!list_empty(&con->rsp_wr_wait_list)) {
1185                 struct rtrs_srv_op *id;
1186                 int ret;
1187
1188                 id = list_entry(con->rsp_wr_wait_list.next,
1189                                 struct rtrs_srv_op, wait_list);
1190                 list_del(&id->wait_list);
1191
1192                 spin_unlock(&con->rsp_wr_wait_lock);
1193                 ret = rtrs_srv_resp_rdma(id, id->status);
1194                 spin_lock(&con->rsp_wr_wait_lock);
1195
1196                 if (!ret) {
1197                         list_add(&id->wait_list, &con->rsp_wr_wait_list);
1198                         break;
1199                 }
1200         }
1201         spin_unlock(&con->rsp_wr_wait_lock);
1202 }
1203
1204 static void rtrs_srv_rdma_done(struct ib_cq *cq, struct ib_wc *wc)
1205 {
1206         struct rtrs_srv_con *con = to_srv_con(wc->qp->qp_context);
1207         struct rtrs_path *s = con->c.path;
1208         struct rtrs_srv_path *srv_path = to_srv_path(s);
1209         struct rtrs_srv_sess *srv = srv_path->srv;
1210         u32 imm_type, imm_payload;
1211         int err;
1212
1213         if (wc->status != IB_WC_SUCCESS) {
1214                 if (wc->status != IB_WC_WR_FLUSH_ERR) {
1215                         rtrs_err(s,
1216                                   "%s (wr_cqe: %p, type: %d, vendor_err: 0x%x, len: %u)\n",
1217                                   ib_wc_status_msg(wc->status), wc->wr_cqe,
1218                                   wc->opcode, wc->vendor_err, wc->byte_len);
1219                         close_path(srv_path);
1220                 }
1221                 return;
1222         }
1223
1224         switch (wc->opcode) {
1225         case IB_WC_RECV_RDMA_WITH_IMM:
1226                 /*
1227                  * post_recv() RDMA write completions of IO reqs (read/write)
1228                  * and hb
1229                  */
1230                 if (WARN_ON(wc->wr_cqe != &io_comp_cqe))
1231                         return;
1232                 err = rtrs_post_recv_empty(&con->c, &io_comp_cqe);
1233                 if (err) {
1234                         rtrs_err(s, "rtrs_post_recv(), err: %d\n", err);
1235                         close_path(srv_path);
1236                         break;
1237                 }
1238                 rtrs_from_imm(be32_to_cpu(wc->ex.imm_data),
1239                                &imm_type, &imm_payload);
1240                 if (imm_type == RTRS_IO_REQ_IMM) {
1241                         u32 msg_id, off;
1242                         void *data;
1243
1244                         msg_id = imm_payload >> srv_path->mem_bits;
1245                         off = imm_payload & ((1 << srv_path->mem_bits) - 1);
1246                         if (msg_id >= srv->queue_depth || off >= max_chunk_size) {
1247                                 rtrs_err(s, "Wrong msg_id %u, off %u\n",
1248                                           msg_id, off);
1249                                 close_path(srv_path);
1250                                 return;
1251                         }
1252                         if (always_invalidate) {
1253                                 struct rtrs_srv_mr *mr = &srv_path->mrs[msg_id];
1254
1255                                 mr->msg_off = off;
1256                                 mr->msg_id = msg_id;
1257                                 err = rtrs_srv_inv_rkey(con, mr);
1258                                 if (err) {
1259                                         rtrs_err(s, "rtrs_post_recv(), err: %d\n",
1260                                                   err);
1261                                         close_path(srv_path);
1262                                         break;
1263                                 }
1264                         } else {
1265                                 data = page_address(srv->chunks[msg_id]) + off;
1266                                 process_io_req(con, data, msg_id, off);
1267                         }
1268                 } else if (imm_type == RTRS_HB_MSG_IMM) {
1269                         WARN_ON(con->c.cid);
1270                         rtrs_send_hb_ack(&srv_path->s);
1271                 } else if (imm_type == RTRS_HB_ACK_IMM) {
1272                         WARN_ON(con->c.cid);
1273                         srv_path->s.hb_missed_cnt = 0;
1274                 } else {
1275                         rtrs_wrn(s, "Unknown IMM type %u\n", imm_type);
1276                 }
1277                 break;
1278         case IB_WC_RDMA_WRITE:
1279         case IB_WC_SEND:
1280                 /*
1281                  * post_send() RDMA write completions of IO reqs (read/write)
1282                  * and hb.
1283                  */
1284                 atomic_add(s->signal_interval, &con->c.sq_wr_avail);
1285
1286                 if (!list_empty_careful(&con->rsp_wr_wait_list))
1287                         rtrs_rdma_process_wr_wait_list(con);
1288
1289                 break;
1290         default:
1291                 rtrs_wrn(s, "Unexpected WC type: %d\n", wc->opcode);
1292                 return;
1293         }
1294 }
1295
1296 /**
1297  * rtrs_srv_get_path_name() - Get rtrs_srv peer hostname.
1298  * @srv:        Session
1299  * @pathname:   Pathname buffer
1300  * @len:        Length of sessname buffer
1301  */
1302 int rtrs_srv_get_path_name(struct rtrs_srv_sess *srv, char *pathname,
1303                            size_t len)
1304 {
1305         struct rtrs_srv_path *srv_path;
1306         int err = -ENOTCONN;
1307
1308         mutex_lock(&srv->paths_mutex);
1309         list_for_each_entry(srv_path, &srv->paths_list, s.entry) {
1310                 if (srv_path->state != RTRS_SRV_CONNECTED)
1311                         continue;
1312                 strscpy(pathname, srv_path->s.sessname,
1313                         min_t(size_t, sizeof(srv_path->s.sessname), len));
1314                 err = 0;
1315                 break;
1316         }
1317         mutex_unlock(&srv->paths_mutex);
1318
1319         return err;
1320 }
1321 EXPORT_SYMBOL(rtrs_srv_get_path_name);
1322
1323 /**
1324  * rtrs_srv_get_queue_depth() - Get rtrs_srv qdepth.
1325  * @srv:        Session
1326  */
1327 int rtrs_srv_get_queue_depth(struct rtrs_srv_sess *srv)
1328 {
1329         return srv->queue_depth;
1330 }
1331 EXPORT_SYMBOL(rtrs_srv_get_queue_depth);
1332
1333 static int find_next_bit_ring(struct rtrs_srv_path *srv_path)
1334 {
1335         struct ib_device *ib_dev = srv_path->s.dev->ib_dev;
1336         int v;
1337
1338         v = cpumask_next(srv_path->cur_cq_vector, &cq_affinity_mask);
1339         if (v >= nr_cpu_ids || v >= ib_dev->num_comp_vectors)
1340                 v = cpumask_first(&cq_affinity_mask);
1341         return v;
1342 }
1343
1344 static int rtrs_srv_get_next_cq_vector(struct rtrs_srv_path *srv_path)
1345 {
1346         srv_path->cur_cq_vector = find_next_bit_ring(srv_path);
1347
1348         return srv_path->cur_cq_vector;
1349 }
1350
1351 static void rtrs_srv_dev_release(struct device *dev)
1352 {
1353         struct rtrs_srv_sess *srv = container_of(dev, struct rtrs_srv_sess,
1354                                                  dev);
1355
1356         kfree(srv);
1357 }
1358
1359 static void free_srv(struct rtrs_srv_sess *srv)
1360 {
1361         int i;
1362
1363         WARN_ON(refcount_read(&srv->refcount));
1364         for (i = 0; i < srv->queue_depth; i++)
1365                 __free_pages(srv->chunks[i], get_order(max_chunk_size));
1366         kfree(srv->chunks);
1367         mutex_destroy(&srv->paths_mutex);
1368         mutex_destroy(&srv->paths_ev_mutex);
1369         /* last put to release the srv structure */
1370         put_device(&srv->dev);
1371 }
1372
1373 static struct rtrs_srv_sess *get_or_create_srv(struct rtrs_srv_ctx *ctx,
1374                                           const uuid_t *paths_uuid,
1375                                           bool first_conn)
1376 {
1377         struct rtrs_srv_sess *srv;
1378         int i;
1379
1380         mutex_lock(&ctx->srv_mutex);
1381         list_for_each_entry(srv, &ctx->srv_list, ctx_list) {
1382                 if (uuid_equal(&srv->paths_uuid, paths_uuid) &&
1383                     refcount_inc_not_zero(&srv->refcount)) {
1384                         mutex_unlock(&ctx->srv_mutex);
1385                         return srv;
1386                 }
1387         }
1388         mutex_unlock(&ctx->srv_mutex);
1389         /*
1390          * If this request is not the first connection request from the
1391          * client for this session then fail and return error.
1392          */
1393         if (!first_conn) {
1394                 pr_err_ratelimited("Error: Not the first connection request for this session\n");
1395                 return ERR_PTR(-ENXIO);
1396         }
1397
1398         /* need to allocate a new srv */
1399         srv = kzalloc(sizeof(*srv), GFP_KERNEL);
1400         if  (!srv)
1401                 return ERR_PTR(-ENOMEM);
1402
1403         INIT_LIST_HEAD(&srv->paths_list);
1404         mutex_init(&srv->paths_mutex);
1405         mutex_init(&srv->paths_ev_mutex);
1406         uuid_copy(&srv->paths_uuid, paths_uuid);
1407         srv->queue_depth = sess_queue_depth;
1408         srv->ctx = ctx;
1409         device_initialize(&srv->dev);
1410         srv->dev.release = rtrs_srv_dev_release;
1411
1412         srv->chunks = kcalloc(srv->queue_depth, sizeof(*srv->chunks),
1413                               GFP_KERNEL);
1414         if (!srv->chunks)
1415                 goto err_free_srv;
1416
1417         for (i = 0; i < srv->queue_depth; i++) {
1418                 srv->chunks[i] = alloc_pages(GFP_KERNEL,
1419                                              get_order(max_chunk_size));
1420                 if (!srv->chunks[i])
1421                         goto err_free_chunks;
1422         }
1423         refcount_set(&srv->refcount, 1);
1424         mutex_lock(&ctx->srv_mutex);
1425         list_add(&srv->ctx_list, &ctx->srv_list);
1426         mutex_unlock(&ctx->srv_mutex);
1427
1428         return srv;
1429
1430 err_free_chunks:
1431         while (i--)
1432                 __free_pages(srv->chunks[i], get_order(max_chunk_size));
1433         kfree(srv->chunks);
1434
1435 err_free_srv:
1436         kfree(srv);
1437         return ERR_PTR(-ENOMEM);
1438 }
1439
1440 static void put_srv(struct rtrs_srv_sess *srv)
1441 {
1442         if (refcount_dec_and_test(&srv->refcount)) {
1443                 struct rtrs_srv_ctx *ctx = srv->ctx;
1444
1445                 WARN_ON(srv->dev.kobj.state_in_sysfs);
1446
1447                 mutex_lock(&ctx->srv_mutex);
1448                 list_del(&srv->ctx_list);
1449                 mutex_unlock(&ctx->srv_mutex);
1450                 free_srv(srv);
1451         }
1452 }
1453
1454 static void __add_path_to_srv(struct rtrs_srv_sess *srv,
1455                               struct rtrs_srv_path *srv_path)
1456 {
1457         list_add_tail(&srv_path->s.entry, &srv->paths_list);
1458         srv->paths_num++;
1459         WARN_ON(srv->paths_num >= MAX_PATHS_NUM);
1460 }
1461
1462 static void del_path_from_srv(struct rtrs_srv_path *srv_path)
1463 {
1464         struct rtrs_srv_sess *srv = srv_path->srv;
1465
1466         if (WARN_ON(!srv))
1467                 return;
1468
1469         mutex_lock(&srv->paths_mutex);
1470         list_del(&srv_path->s.entry);
1471         WARN_ON(!srv->paths_num);
1472         srv->paths_num--;
1473         mutex_unlock(&srv->paths_mutex);
1474 }
1475
1476 /* return true if addresses are the same, error other wise */
1477 static int sockaddr_cmp(const struct sockaddr *a, const struct sockaddr *b)
1478 {
1479         switch (a->sa_family) {
1480         case AF_IB:
1481                 return memcmp(&((struct sockaddr_ib *)a)->sib_addr,
1482                               &((struct sockaddr_ib *)b)->sib_addr,
1483                               sizeof(struct ib_addr)) &&
1484                         (b->sa_family == AF_IB);
1485         case AF_INET:
1486                 return memcmp(&((struct sockaddr_in *)a)->sin_addr,
1487                               &((struct sockaddr_in *)b)->sin_addr,
1488                               sizeof(struct in_addr)) &&
1489                         (b->sa_family == AF_INET);
1490         case AF_INET6:
1491                 return memcmp(&((struct sockaddr_in6 *)a)->sin6_addr,
1492                               &((struct sockaddr_in6 *)b)->sin6_addr,
1493                               sizeof(struct in6_addr)) &&
1494                         (b->sa_family == AF_INET6);
1495         default:
1496                 return -ENOENT;
1497         }
1498 }
1499
1500 static bool __is_path_w_addr_exists(struct rtrs_srv_sess *srv,
1501                                     struct rdma_addr *addr)
1502 {
1503         struct rtrs_srv_path *srv_path;
1504
1505         list_for_each_entry(srv_path, &srv->paths_list, s.entry)
1506                 if (!sockaddr_cmp((struct sockaddr *)&srv_path->s.dst_addr,
1507                                   (struct sockaddr *)&addr->dst_addr) &&
1508                     !sockaddr_cmp((struct sockaddr *)&srv_path->s.src_addr,
1509                                   (struct sockaddr *)&addr->src_addr))
1510                         return true;
1511
1512         return false;
1513 }
1514
1515 static void free_path(struct rtrs_srv_path *srv_path)
1516 {
1517         if (srv_path->kobj.state_in_sysfs) {
1518                 kobject_del(&srv_path->kobj);
1519                 kobject_put(&srv_path->kobj);
1520         } else {
1521                 free_percpu(srv_path->stats->rdma_stats);
1522                 kfree(srv_path->stats);
1523                 kfree(srv_path);
1524         }
1525 }
1526
1527 static void rtrs_srv_close_work(struct work_struct *work)
1528 {
1529         struct rtrs_srv_path *srv_path;
1530         struct rtrs_srv_con *con;
1531         int i;
1532
1533         srv_path = container_of(work, typeof(*srv_path), close_work);
1534
1535         rtrs_srv_stop_hb(srv_path);
1536
1537         for (i = 0; i < srv_path->s.con_num; i++) {
1538                 if (!srv_path->s.con[i])
1539                         continue;
1540                 con = to_srv_con(srv_path->s.con[i]);
1541                 rdma_disconnect(con->c.cm_id);
1542                 ib_drain_qp(con->c.qp);
1543         }
1544
1545         /*
1546          * Degrade ref count to the usual model with a single shared
1547          * atomic_t counter
1548          */
1549         percpu_ref_kill(&srv_path->ids_inflight_ref);
1550
1551         /* Wait for all completion */
1552         wait_for_completion(&srv_path->complete_done);
1553
1554         rtrs_srv_destroy_path_files(srv_path);
1555
1556         /* Notify upper layer if we are the last path */
1557         rtrs_srv_path_down(srv_path);
1558
1559         unmap_cont_bufs(srv_path);
1560         rtrs_srv_free_ops_ids(srv_path);
1561
1562         for (i = 0; i < srv_path->s.con_num; i++) {
1563                 if (!srv_path->s.con[i])
1564                         continue;
1565                 con = to_srv_con(srv_path->s.con[i]);
1566                 rtrs_cq_qp_destroy(&con->c);
1567                 rdma_destroy_id(con->c.cm_id);
1568                 kfree(con);
1569         }
1570         rtrs_ib_dev_put(srv_path->s.dev);
1571
1572         del_path_from_srv(srv_path);
1573         put_srv(srv_path->srv);
1574         srv_path->srv = NULL;
1575         rtrs_srv_change_state(srv_path, RTRS_SRV_CLOSED);
1576
1577         kfree(srv_path->dma_addr);
1578         kfree(srv_path->s.con);
1579         free_path(srv_path);
1580 }
1581
1582 static int rtrs_rdma_do_accept(struct rtrs_srv_path *srv_path,
1583                                struct rdma_cm_id *cm_id)
1584 {
1585         struct rtrs_srv_sess *srv = srv_path->srv;
1586         struct rtrs_msg_conn_rsp msg;
1587         struct rdma_conn_param param;
1588         int err;
1589
1590         param = (struct rdma_conn_param) {
1591                 .rnr_retry_count = 7,
1592                 .private_data = &msg,
1593                 .private_data_len = sizeof(msg),
1594         };
1595
1596         msg = (struct rtrs_msg_conn_rsp) {
1597                 .magic = cpu_to_le16(RTRS_MAGIC),
1598                 .version = cpu_to_le16(RTRS_PROTO_VER),
1599                 .queue_depth = cpu_to_le16(srv->queue_depth),
1600                 .max_io_size = cpu_to_le32(max_chunk_size - MAX_HDR_SIZE),
1601                 .max_hdr_size = cpu_to_le32(MAX_HDR_SIZE),
1602         };
1603
1604         if (always_invalidate)
1605                 msg.flags = cpu_to_le32(RTRS_MSG_NEW_RKEY_F);
1606
1607         err = rdma_accept(cm_id, &param);
1608         if (err)
1609                 pr_err("rdma_accept(), err: %d\n", err);
1610
1611         return err;
1612 }
1613
1614 static int rtrs_rdma_do_reject(struct rdma_cm_id *cm_id, int errno)
1615 {
1616         struct rtrs_msg_conn_rsp msg;
1617         int err;
1618
1619         msg = (struct rtrs_msg_conn_rsp) {
1620                 .magic = cpu_to_le16(RTRS_MAGIC),
1621                 .version = cpu_to_le16(RTRS_PROTO_VER),
1622                 .errno = cpu_to_le16(errno),
1623         };
1624
1625         err = rdma_reject(cm_id, &msg, sizeof(msg), IB_CM_REJ_CONSUMER_DEFINED);
1626         if (err)
1627                 pr_err("rdma_reject(), err: %d\n", err);
1628
1629         /* Bounce errno back */
1630         return errno;
1631 }
1632
1633 static struct rtrs_srv_path *
1634 __find_path(struct rtrs_srv_sess *srv, const uuid_t *sess_uuid)
1635 {
1636         struct rtrs_srv_path *srv_path;
1637
1638         list_for_each_entry(srv_path, &srv->paths_list, s.entry) {
1639                 if (uuid_equal(&srv_path->s.uuid, sess_uuid))
1640                         return srv_path;
1641         }
1642
1643         return NULL;
1644 }
1645
1646 static int create_con(struct rtrs_srv_path *srv_path,
1647                       struct rdma_cm_id *cm_id,
1648                       unsigned int cid)
1649 {
1650         struct rtrs_srv_sess *srv = srv_path->srv;
1651         struct rtrs_path *s = &srv_path->s;
1652         struct rtrs_srv_con *con;
1653
1654         u32 cq_num, max_send_wr, max_recv_wr, wr_limit;
1655         int err, cq_vector;
1656
1657         con = kzalloc(sizeof(*con), GFP_KERNEL);
1658         if (!con) {
1659                 err = -ENOMEM;
1660                 goto err;
1661         }
1662
1663         spin_lock_init(&con->rsp_wr_wait_lock);
1664         INIT_LIST_HEAD(&con->rsp_wr_wait_list);
1665         con->c.cm_id = cm_id;
1666         con->c.path = &srv_path->s;
1667         con->c.cid = cid;
1668         atomic_set(&con->c.wr_cnt, 1);
1669         wr_limit = srv_path->s.dev->ib_dev->attrs.max_qp_wr;
1670
1671         if (con->c.cid == 0) {
1672                 /*
1673                  * All receive and all send (each requiring invalidate)
1674                  * + 2 for drain and heartbeat
1675                  */
1676                 max_send_wr = min_t(int, wr_limit,
1677                                     SERVICE_CON_QUEUE_DEPTH * 2 + 2);
1678                 max_recv_wr = max_send_wr;
1679                 s->signal_interval = min_not_zero(srv->queue_depth,
1680                                                   (size_t)SERVICE_CON_QUEUE_DEPTH);
1681         } else {
1682                 /* when always_invlaidate enalbed, we need linv+rinv+mr+imm */
1683                 if (always_invalidate)
1684                         max_send_wr =
1685                                 min_t(int, wr_limit,
1686                                       srv->queue_depth * (1 + 4) + 1);
1687                 else
1688                         max_send_wr =
1689                                 min_t(int, wr_limit,
1690                                       srv->queue_depth * (1 + 2) + 1);
1691
1692                 max_recv_wr = srv->queue_depth + 1;
1693         }
1694         cq_num = max_send_wr + max_recv_wr;
1695         atomic_set(&con->c.sq_wr_avail, max_send_wr);
1696         cq_vector = rtrs_srv_get_next_cq_vector(srv_path);
1697
1698         /* TODO: SOFTIRQ can be faster, but be careful with softirq context */
1699         err = rtrs_cq_qp_create(&srv_path->s, &con->c, 1, cq_vector, cq_num,
1700                                  max_send_wr, max_recv_wr,
1701                                  IB_POLL_WORKQUEUE);
1702         if (err) {
1703                 rtrs_err(s, "rtrs_cq_qp_create(), err: %d\n", err);
1704                 goto free_con;
1705         }
1706         if (con->c.cid == 0) {
1707                 err = post_recv_info_req(con);
1708                 if (err)
1709                         goto free_cqqp;
1710         }
1711         WARN_ON(srv_path->s.con[cid]);
1712         srv_path->s.con[cid] = &con->c;
1713
1714         /*
1715          * Change context from server to current connection.  The other
1716          * way is to use cm_id->qp->qp_context, which does not work on OFED.
1717          */
1718         cm_id->context = &con->c;
1719
1720         return 0;
1721
1722 free_cqqp:
1723         rtrs_cq_qp_destroy(&con->c);
1724 free_con:
1725         kfree(con);
1726
1727 err:
1728         return err;
1729 }
1730
1731 static struct rtrs_srv_path *__alloc_path(struct rtrs_srv_sess *srv,
1732                                            struct rdma_cm_id *cm_id,
1733                                            unsigned int con_num,
1734                                            unsigned int recon_cnt,
1735                                            const uuid_t *uuid)
1736 {
1737         struct rtrs_srv_path *srv_path;
1738         int err = -ENOMEM;
1739         char str[NAME_MAX];
1740         struct rtrs_addr path;
1741
1742         if (srv->paths_num >= MAX_PATHS_NUM) {
1743                 err = -ECONNRESET;
1744                 goto err;
1745         }
1746         if (__is_path_w_addr_exists(srv, &cm_id->route.addr)) {
1747                 err = -EEXIST;
1748                 pr_err("Path with same addr exists\n");
1749                 goto err;
1750         }
1751         srv_path = kzalloc(sizeof(*srv_path), GFP_KERNEL);
1752         if (!srv_path)
1753                 goto err;
1754
1755         srv_path->stats = kzalloc(sizeof(*srv_path->stats), GFP_KERNEL);
1756         if (!srv_path->stats)
1757                 goto err_free_sess;
1758
1759         srv_path->stats->rdma_stats = alloc_percpu(struct rtrs_srv_stats_rdma_stats);
1760         if (!srv_path->stats->rdma_stats)
1761                 goto err_free_stats;
1762
1763         srv_path->stats->srv_path = srv_path;
1764
1765         srv_path->dma_addr = kcalloc(srv->queue_depth,
1766                                      sizeof(*srv_path->dma_addr),
1767                                      GFP_KERNEL);
1768         if (!srv_path->dma_addr)
1769                 goto err_free_percpu;
1770
1771         srv_path->s.con = kcalloc(con_num, sizeof(*srv_path->s.con),
1772                                   GFP_KERNEL);
1773         if (!srv_path->s.con)
1774                 goto err_free_dma_addr;
1775
1776         srv_path->state = RTRS_SRV_CONNECTING;
1777         srv_path->srv = srv;
1778         srv_path->cur_cq_vector = -1;
1779         srv_path->s.dst_addr = cm_id->route.addr.dst_addr;
1780         srv_path->s.src_addr = cm_id->route.addr.src_addr;
1781
1782         /* temporary until receiving session-name from client */
1783         path.src = &srv_path->s.src_addr;
1784         path.dst = &srv_path->s.dst_addr;
1785         rtrs_addr_to_str(&path, str, sizeof(str));
1786         strscpy(srv_path->s.sessname, str, sizeof(srv_path->s.sessname));
1787
1788         srv_path->s.con_num = con_num;
1789         srv_path->s.irq_con_num = con_num;
1790         srv_path->s.recon_cnt = recon_cnt;
1791         uuid_copy(&srv_path->s.uuid, uuid);
1792         spin_lock_init(&srv_path->state_lock);
1793         INIT_WORK(&srv_path->close_work, rtrs_srv_close_work);
1794         rtrs_srv_init_hb(srv_path);
1795
1796         srv_path->s.dev = rtrs_ib_dev_find_or_add(cm_id->device, &dev_pd);
1797         if (!srv_path->s.dev) {
1798                 err = -ENOMEM;
1799                 goto err_free_con;
1800         }
1801         err = map_cont_bufs(srv_path);
1802         if (err)
1803                 goto err_put_dev;
1804
1805         err = rtrs_srv_alloc_ops_ids(srv_path);
1806         if (err)
1807                 goto err_unmap_bufs;
1808
1809         __add_path_to_srv(srv, srv_path);
1810
1811         return srv_path;
1812
1813 err_unmap_bufs:
1814         unmap_cont_bufs(srv_path);
1815 err_put_dev:
1816         rtrs_ib_dev_put(srv_path->s.dev);
1817 err_free_con:
1818         kfree(srv_path->s.con);
1819 err_free_dma_addr:
1820         kfree(srv_path->dma_addr);
1821 err_free_percpu:
1822         free_percpu(srv_path->stats->rdma_stats);
1823 err_free_stats:
1824         kfree(srv_path->stats);
1825 err_free_sess:
1826         kfree(srv_path);
1827 err:
1828         return ERR_PTR(err);
1829 }
1830
1831 static int rtrs_rdma_connect(struct rdma_cm_id *cm_id,
1832                               const struct rtrs_msg_conn_req *msg,
1833                               size_t len)
1834 {
1835         struct rtrs_srv_ctx *ctx = cm_id->context;
1836         struct rtrs_srv_path *srv_path;
1837         struct rtrs_srv_sess *srv;
1838
1839         u16 version, con_num, cid;
1840         u16 recon_cnt;
1841         int err = -ECONNRESET;
1842
1843         if (len < sizeof(*msg)) {
1844                 pr_err("Invalid RTRS connection request\n");
1845                 goto reject_w_err;
1846         }
1847         if (le16_to_cpu(msg->magic) != RTRS_MAGIC) {
1848                 pr_err("Invalid RTRS magic\n");
1849                 goto reject_w_err;
1850         }
1851         version = le16_to_cpu(msg->version);
1852         if (version >> 8 != RTRS_PROTO_VER_MAJOR) {
1853                 pr_err("Unsupported major RTRS version: %d, expected %d\n",
1854                        version >> 8, RTRS_PROTO_VER_MAJOR);
1855                 goto reject_w_err;
1856         }
1857         con_num = le16_to_cpu(msg->cid_num);
1858         if (con_num > 4096) {
1859                 /* Sanity check */
1860                 pr_err("Too many connections requested: %d\n", con_num);
1861                 goto reject_w_err;
1862         }
1863         cid = le16_to_cpu(msg->cid);
1864         if (cid >= con_num) {
1865                 /* Sanity check */
1866                 pr_err("Incorrect cid: %d >= %d\n", cid, con_num);
1867                 goto reject_w_err;
1868         }
1869         recon_cnt = le16_to_cpu(msg->recon_cnt);
1870         srv = get_or_create_srv(ctx, &msg->paths_uuid, msg->first_conn);
1871         if (IS_ERR(srv)) {
1872                 err = PTR_ERR(srv);
1873                 pr_err("get_or_create_srv(), error %d\n", err);
1874                 goto reject_w_err;
1875         }
1876         mutex_lock(&srv->paths_mutex);
1877         srv_path = __find_path(srv, &msg->sess_uuid);
1878         if (srv_path) {
1879                 struct rtrs_path *s = &srv_path->s;
1880
1881                 /* Session already holds a reference */
1882                 put_srv(srv);
1883
1884                 if (srv_path->state != RTRS_SRV_CONNECTING) {
1885                         rtrs_err(s, "Session in wrong state: %s\n",
1886                                   rtrs_srv_state_str(srv_path->state));
1887                         mutex_unlock(&srv->paths_mutex);
1888                         goto reject_w_err;
1889                 }
1890                 /*
1891                  * Sanity checks
1892                  */
1893                 if (con_num != s->con_num || cid >= s->con_num) {
1894                         rtrs_err(s, "Incorrect request: %d, %d\n",
1895                                   cid, con_num);
1896                         mutex_unlock(&srv->paths_mutex);
1897                         goto reject_w_err;
1898                 }
1899                 if (s->con[cid]) {
1900                         rtrs_err(s, "Connection already exists: %d\n",
1901                                   cid);
1902                         mutex_unlock(&srv->paths_mutex);
1903                         goto reject_w_err;
1904                 }
1905         } else {
1906                 srv_path = __alloc_path(srv, cm_id, con_num, recon_cnt,
1907                                     &msg->sess_uuid);
1908                 if (IS_ERR(srv_path)) {
1909                         mutex_unlock(&srv->paths_mutex);
1910                         put_srv(srv);
1911                         err = PTR_ERR(srv_path);
1912                         pr_err("RTRS server session allocation failed: %d\n", err);
1913                         goto reject_w_err;
1914                 }
1915         }
1916         err = create_con(srv_path, cm_id, cid);
1917         if (err) {
1918                 rtrs_err((&srv_path->s), "create_con(), error %d\n", err);
1919                 rtrs_rdma_do_reject(cm_id, err);
1920                 /*
1921                  * Since session has other connections we follow normal way
1922                  * through workqueue, but still return an error to tell cma.c
1923                  * to call rdma_destroy_id() for current connection.
1924                  */
1925                 goto close_and_return_err;
1926         }
1927         err = rtrs_rdma_do_accept(srv_path, cm_id);
1928         if (err) {
1929                 rtrs_err((&srv_path->s), "rtrs_rdma_do_accept(), error %d\n", err);
1930                 rtrs_rdma_do_reject(cm_id, err);
1931                 /*
1932                  * Since current connection was successfully added to the
1933                  * session we follow normal way through workqueue to close the
1934                  * session, thus return 0 to tell cma.c we call
1935                  * rdma_destroy_id() ourselves.
1936                  */
1937                 err = 0;
1938                 goto close_and_return_err;
1939         }
1940         mutex_unlock(&srv->paths_mutex);
1941
1942         return 0;
1943
1944 reject_w_err:
1945         return rtrs_rdma_do_reject(cm_id, err);
1946
1947 close_and_return_err:
1948         mutex_unlock(&srv->paths_mutex);
1949         close_path(srv_path);
1950
1951         return err;
1952 }
1953
1954 static int rtrs_srv_rdma_cm_handler(struct rdma_cm_id *cm_id,
1955                                      struct rdma_cm_event *ev)
1956 {
1957         struct rtrs_srv_path *srv_path = NULL;
1958         struct rtrs_path *s = NULL;
1959         struct rtrs_con *c = NULL;
1960
1961         if (ev->event == RDMA_CM_EVENT_CONNECT_REQUEST)
1962                 /*
1963                  * In case of error cma.c will destroy cm_id,
1964                  * see cma_process_remove()
1965                  */
1966                 return rtrs_rdma_connect(cm_id, ev->param.conn.private_data,
1967                                           ev->param.conn.private_data_len);
1968
1969         c = cm_id->context;
1970         s = c->path;
1971         srv_path = to_srv_path(s);
1972
1973         switch (ev->event) {
1974         case RDMA_CM_EVENT_ESTABLISHED:
1975                 /* Nothing here */
1976                 break;
1977         case RDMA_CM_EVENT_REJECTED:
1978         case RDMA_CM_EVENT_CONNECT_ERROR:
1979         case RDMA_CM_EVENT_UNREACHABLE:
1980                 rtrs_err(s, "CM error (CM event: %s, err: %d)\n",
1981                           rdma_event_msg(ev->event), ev->status);
1982                 fallthrough;
1983         case RDMA_CM_EVENT_DISCONNECTED:
1984         case RDMA_CM_EVENT_ADDR_CHANGE:
1985         case RDMA_CM_EVENT_TIMEWAIT_EXIT:
1986         case RDMA_CM_EVENT_DEVICE_REMOVAL:
1987                 close_path(srv_path);
1988                 break;
1989         default:
1990                 pr_err("Ignoring unexpected CM event %s, err %d\n",
1991                        rdma_event_msg(ev->event), ev->status);
1992                 break;
1993         }
1994
1995         return 0;
1996 }
1997
1998 static struct rdma_cm_id *rtrs_srv_cm_init(struct rtrs_srv_ctx *ctx,
1999                                             struct sockaddr *addr,
2000                                             enum rdma_ucm_port_space ps)
2001 {
2002         struct rdma_cm_id *cm_id;
2003         int ret;
2004
2005         cm_id = rdma_create_id(&init_net, rtrs_srv_rdma_cm_handler,
2006                                ctx, ps, IB_QPT_RC);
2007         if (IS_ERR(cm_id)) {
2008                 ret = PTR_ERR(cm_id);
2009                 pr_err("Creating id for RDMA connection failed, err: %d\n",
2010                        ret);
2011                 goto err_out;
2012         }
2013         ret = rdma_bind_addr(cm_id, addr);
2014         if (ret) {
2015                 pr_err("Binding RDMA address failed, err: %d\n", ret);
2016                 goto err_cm;
2017         }
2018         ret = rdma_listen(cm_id, 64);
2019         if (ret) {
2020                 pr_err("Listening on RDMA connection failed, err: %d\n",
2021                        ret);
2022                 goto err_cm;
2023         }
2024
2025         return cm_id;
2026
2027 err_cm:
2028         rdma_destroy_id(cm_id);
2029 err_out:
2030
2031         return ERR_PTR(ret);
2032 }
2033
2034 static int rtrs_srv_rdma_init(struct rtrs_srv_ctx *ctx, u16 port)
2035 {
2036         struct sockaddr_in6 sin = {
2037                 .sin6_family    = AF_INET6,
2038                 .sin6_addr      = IN6ADDR_ANY_INIT,
2039                 .sin6_port      = htons(port),
2040         };
2041         struct sockaddr_ib sib = {
2042                 .sib_family                     = AF_IB,
2043                 .sib_sid        = cpu_to_be64(RDMA_IB_IP_PS_IB | port),
2044                 .sib_sid_mask   = cpu_to_be64(0xffffffffffffffffULL),
2045                 .sib_pkey       = cpu_to_be16(0xffff),
2046         };
2047         struct rdma_cm_id *cm_ip, *cm_ib;
2048         int ret;
2049
2050         /*
2051          * We accept both IPoIB and IB connections, so we need to keep
2052          * two cm id's, one for each socket type and port space.
2053          * If the cm initialization of one of the id's fails, we abort
2054          * everything.
2055          */
2056         cm_ip = rtrs_srv_cm_init(ctx, (struct sockaddr *)&sin, RDMA_PS_TCP);
2057         if (IS_ERR(cm_ip))
2058                 return PTR_ERR(cm_ip);
2059
2060         cm_ib = rtrs_srv_cm_init(ctx, (struct sockaddr *)&sib, RDMA_PS_IB);
2061         if (IS_ERR(cm_ib)) {
2062                 ret = PTR_ERR(cm_ib);
2063                 goto free_cm_ip;
2064         }
2065
2066         ctx->cm_id_ip = cm_ip;
2067         ctx->cm_id_ib = cm_ib;
2068
2069         return 0;
2070
2071 free_cm_ip:
2072         rdma_destroy_id(cm_ip);
2073
2074         return ret;
2075 }
2076
2077 static struct rtrs_srv_ctx *alloc_srv_ctx(struct rtrs_srv_ops *ops)
2078 {
2079         struct rtrs_srv_ctx *ctx;
2080
2081         ctx = kzalloc(sizeof(*ctx), GFP_KERNEL);
2082         if (!ctx)
2083                 return NULL;
2084
2085         ctx->ops = *ops;
2086         mutex_init(&ctx->srv_mutex);
2087         INIT_LIST_HEAD(&ctx->srv_list);
2088
2089         return ctx;
2090 }
2091
2092 static void free_srv_ctx(struct rtrs_srv_ctx *ctx)
2093 {
2094         WARN_ON(!list_empty(&ctx->srv_list));
2095         mutex_destroy(&ctx->srv_mutex);
2096         kfree(ctx);
2097 }
2098
2099 static int rtrs_srv_add_one(struct ib_device *device)
2100 {
2101         struct rtrs_srv_ctx *ctx;
2102         int ret = 0;
2103
2104         mutex_lock(&ib_ctx.ib_dev_mutex);
2105         if (ib_ctx.ib_dev_count)
2106                 goto out;
2107
2108         /*
2109          * Since our CM IDs are NOT bound to any ib device we will create them
2110          * only once
2111          */
2112         ctx = ib_ctx.srv_ctx;
2113         ret = rtrs_srv_rdma_init(ctx, ib_ctx.port);
2114         if (ret) {
2115                 /*
2116                  * We errored out here.
2117                  * According to the ib code, if we encounter an error here then the
2118                  * error code is ignored, and no more calls to our ops are made.
2119                  */
2120                 pr_err("Failed to initialize RDMA connection");
2121                 goto err_out;
2122         }
2123
2124 out:
2125         /*
2126          * Keep a track on the number of ib devices added
2127          */
2128         ib_ctx.ib_dev_count++;
2129
2130 err_out:
2131         mutex_unlock(&ib_ctx.ib_dev_mutex);
2132         return ret;
2133 }
2134
2135 static void rtrs_srv_remove_one(struct ib_device *device, void *client_data)
2136 {
2137         struct rtrs_srv_ctx *ctx;
2138
2139         mutex_lock(&ib_ctx.ib_dev_mutex);
2140         ib_ctx.ib_dev_count--;
2141
2142         if (ib_ctx.ib_dev_count)
2143                 goto out;
2144
2145         /*
2146          * Since our CM IDs are NOT bound to any ib device we will remove them
2147          * only once, when the last device is removed
2148          */
2149         ctx = ib_ctx.srv_ctx;
2150         rdma_destroy_id(ctx->cm_id_ip);
2151         rdma_destroy_id(ctx->cm_id_ib);
2152
2153 out:
2154         mutex_unlock(&ib_ctx.ib_dev_mutex);
2155 }
2156
2157 static struct ib_client rtrs_srv_client = {
2158         .name   = "rtrs_server",
2159         .add    = rtrs_srv_add_one,
2160         .remove = rtrs_srv_remove_one
2161 };
2162
2163 /**
2164  * rtrs_srv_open() - open RTRS server context
2165  * @ops:                callback functions
2166  * @port:               port to listen on
2167  *
2168  * Creates server context with specified callbacks.
2169  *
2170  * Return a valid pointer on success otherwise PTR_ERR.
2171  */
2172 struct rtrs_srv_ctx *rtrs_srv_open(struct rtrs_srv_ops *ops, u16 port)
2173 {
2174         struct rtrs_srv_ctx *ctx;
2175         int err;
2176
2177         ctx = alloc_srv_ctx(ops);
2178         if (!ctx)
2179                 return ERR_PTR(-ENOMEM);
2180
2181         mutex_init(&ib_ctx.ib_dev_mutex);
2182         ib_ctx.srv_ctx = ctx;
2183         ib_ctx.port = port;
2184
2185         err = ib_register_client(&rtrs_srv_client);
2186         if (err) {
2187                 free_srv_ctx(ctx);
2188                 return ERR_PTR(err);
2189         }
2190
2191         return ctx;
2192 }
2193 EXPORT_SYMBOL(rtrs_srv_open);
2194
2195 static void close_paths(struct rtrs_srv_sess *srv)
2196 {
2197         struct rtrs_srv_path *srv_path;
2198
2199         mutex_lock(&srv->paths_mutex);
2200         list_for_each_entry(srv_path, &srv->paths_list, s.entry)
2201                 close_path(srv_path);
2202         mutex_unlock(&srv->paths_mutex);
2203 }
2204
2205 static void close_ctx(struct rtrs_srv_ctx *ctx)
2206 {
2207         struct rtrs_srv_sess *srv;
2208
2209         mutex_lock(&ctx->srv_mutex);
2210         list_for_each_entry(srv, &ctx->srv_list, ctx_list)
2211                 close_paths(srv);
2212         mutex_unlock(&ctx->srv_mutex);
2213         flush_workqueue(rtrs_wq);
2214 }
2215
2216 /**
2217  * rtrs_srv_close() - close RTRS server context
2218  * @ctx: pointer to server context
2219  *
2220  * Closes RTRS server context with all client sessions.
2221  */
2222 void rtrs_srv_close(struct rtrs_srv_ctx *ctx)
2223 {
2224         ib_unregister_client(&rtrs_srv_client);
2225         mutex_destroy(&ib_ctx.ib_dev_mutex);
2226         close_ctx(ctx);
2227         free_srv_ctx(ctx);
2228 }
2229 EXPORT_SYMBOL(rtrs_srv_close);
2230
2231 static int check_module_params(void)
2232 {
2233         if (sess_queue_depth < 1 || sess_queue_depth > MAX_SESS_QUEUE_DEPTH) {
2234                 pr_err("Invalid sess_queue_depth value %d, has to be >= %d, <= %d.\n",
2235                        sess_queue_depth, 1, MAX_SESS_QUEUE_DEPTH);
2236                 return -EINVAL;
2237         }
2238         if (max_chunk_size < MIN_CHUNK_SIZE || !is_power_of_2(max_chunk_size)) {
2239                 pr_err("Invalid max_chunk_size value %d, has to be >= %d and should be power of two.\n",
2240                        max_chunk_size, MIN_CHUNK_SIZE);
2241                 return -EINVAL;
2242         }
2243
2244         /*
2245          * Check if IB immediate data size is enough to hold the mem_id and the
2246          * offset inside the memory chunk
2247          */
2248         if ((ilog2(sess_queue_depth - 1) + 1) +
2249             (ilog2(max_chunk_size - 1) + 1) > MAX_IMM_PAYL_BITS) {
2250                 pr_err("RDMA immediate size (%db) not enough to encode %d buffers of size %dB. Reduce 'sess_queue_depth' or 'max_chunk_size' parameters.\n",
2251                        MAX_IMM_PAYL_BITS, sess_queue_depth, max_chunk_size);
2252                 return -EINVAL;
2253         }
2254
2255         return 0;
2256 }
2257
2258 static int __init rtrs_server_init(void)
2259 {
2260         int err;
2261
2262         pr_info("Loading module %s, proto %s: (max_chunk_size: %d (pure IO %ld, headers %ld) , sess_queue_depth: %d, always_invalidate: %d)\n",
2263                 KBUILD_MODNAME, RTRS_PROTO_VER_STRING,
2264                 max_chunk_size, max_chunk_size - MAX_HDR_SIZE, MAX_HDR_SIZE,
2265                 sess_queue_depth, always_invalidate);
2266
2267         rtrs_rdma_dev_pd_init(0, &dev_pd);
2268
2269         err = check_module_params();
2270         if (err) {
2271                 pr_err("Failed to load module, invalid module parameters, err: %d\n",
2272                        err);
2273                 return err;
2274         }
2275         err = class_register(&rtrs_dev_class);
2276         if (err)
2277                 goto out_err;
2278
2279         rtrs_wq = alloc_workqueue("rtrs_server_wq", 0, 0);
2280         if (!rtrs_wq) {
2281                 err = -ENOMEM;
2282                 goto out_dev_class;
2283         }
2284
2285         return 0;
2286
2287 out_dev_class:
2288         class_unregister(&rtrs_dev_class);
2289 out_err:
2290         return err;
2291 }
2292
2293 static void __exit rtrs_server_exit(void)
2294 {
2295         destroy_workqueue(rtrs_wq);
2296         class_unregister(&rtrs_dev_class);
2297         rtrs_rdma_dev_pd_deinit(&dev_pd);
2298 }
2299
2300 module_init(rtrs_server_init);
2301 module_exit(rtrs_server_exit);