]>
git.proxmox.com Git - mirror_ubuntu-artful-kernel.git/blob - net/smc/smc_rx.c
2 * Shared Memory Communications over RDMA (SMC-R) and RoCE
5 * copy new RMBE data into user space
7 * Copyright IBM Corp. 2016
9 * Author(s): Ursula Braun <ubraun@linux.vnet.ibm.com>
12 #include <linux/net.h>
13 #include <linux/rcupdate.h>
19 #include "smc_tx.h" /* smc_tx_consumer_update() */
22 /* callback implementation for sk.sk_data_ready()
23 * to wakeup rcvbuf consumers that blocked with smc_rx_wait_data().
24 * indirectly called by smc_cdc_msg_recv_action().
26 static void smc_rx_data_ready(struct sock
*sk
)
30 /* derived from sock_def_readable() */
31 /* called already in smc_listen_work() */
33 wq
= rcu_dereference(sk
->sk_wq
);
34 if (skwq_has_sleeper(wq
))
35 wake_up_interruptible_sync_poll(&wq
->wait
, POLLIN
| POLLPRI
|
36 POLLRDNORM
| POLLRDBAND
);
37 if ((sk
->sk_shutdown
== SHUTDOWN_MASK
) ||
38 (sk
->sk_state
== SMC_CLOSED
))
39 sk_wake_async(sk
, SOCK_WAKE_WAITD
, POLL_HUP
);
41 sk_wake_async(sk
, SOCK_WAKE_WAITD
, POLL_IN
);
45 /* blocks rcvbuf consumer until >=len bytes available or timeout or interrupted
47 * @timeo pointer to max seconds to wait, pointer to value 0 for no timeout
49 * 1 if at least 1 byte available in rcvbuf or if socket error/shutdown.
50 * 0 otherwise (nothing in rcvbuf nor timeout, e.g. interrupted).
52 static int smc_rx_wait_data(struct smc_sock
*smc
, long *timeo
)
54 DEFINE_WAIT_FUNC(wait
, woken_wake_function
);
55 struct smc_connection
*conn
= &smc
->conn
;
56 struct sock
*sk
= &smc
->sk
;
59 if (atomic_read(&conn
->bytes_to_rcv
))
61 sk_set_bit(SOCKWQ_ASYNC_WAITDATA
, sk
);
62 add_wait_queue(sk_sleep(sk
), &wait
);
63 rc
= sk_wait_event(sk
, timeo
,
65 sk
->sk_shutdown
& RCV_SHUTDOWN
||
66 sock_flag(sk
, SOCK_DONE
) ||
67 atomic_read(&conn
->bytes_to_rcv
) ||
68 smc_cdc_rxed_any_close_or_senddone(conn
),
70 remove_wait_queue(sk_sleep(sk
), &wait
);
71 sk_clear_bit(SOCKWQ_ASYNC_WAITDATA
, sk
);
75 /* rcvbuf consumer: main API called by socket layer.
76 * called under sk lock.
78 int smc_rx_recvmsg(struct smc_sock
*smc
, struct msghdr
*msg
, size_t len
,
81 size_t copylen
, read_done
= 0, read_remaining
= len
;
82 size_t chunk_len
, chunk_off
, chunk_len_sum
;
83 struct smc_connection
*conn
= &smc
->conn
;
84 union smc_host_cursor cons
;
89 int target
; /* Read at least these many bytes */
92 if (unlikely(flags
& MSG_ERRQUEUE
))
93 return -EINVAL
; /* future work for sk.sk_family == AF_SMC */
95 return -EINVAL
; /* future work */
98 if (sk
->sk_state
== SMC_LISTEN
)
100 timeo
= sock_rcvtimeo(sk
, flags
& MSG_DONTWAIT
);
101 target
= sock_rcvlowat(sk
, flags
& MSG_WAITALL
, len
);
103 msg
->msg_namelen
= 0;
104 /* we currently use 1 RMBE per RMB, so RMBE == RMB base addr */
105 rcvbuf_base
= conn
->rmb_desc
->cpu_addr
;
107 do { /* while (read_remaining) */
108 if (read_done
>= target
)
111 if (atomic_read(&conn
->bytes_to_rcv
))
116 sk
->sk_state
== SMC_CLOSED
||
117 (sk
->sk_shutdown
& RCV_SHUTDOWN
) ||
119 signal_pending(current
) ||
120 smc_cdc_rxed_any_close_or_senddone(conn
) ||
121 conn
->local_tx_ctrl
.conn_state_flags
.
125 if (sock_flag(sk
, SOCK_DONE
))
128 read_done
= sock_error(sk
);
131 if (sk
->sk_shutdown
& RCV_SHUTDOWN
||
132 smc_cdc_rxed_any_close_or_senddone(conn
) ||
133 conn
->local_tx_ctrl
.conn_state_flags
.
136 if (sk
->sk_state
== SMC_CLOSED
) {
137 if (!sock_flag(sk
, SOCK_DONE
)) {
138 /* This occurs when user tries to read
139 * from never connected socket.
141 read_done
= -ENOTCONN
;
146 if (signal_pending(current
)) {
147 read_done
= sock_intr_errno(timeo
);
152 if (!atomic_read(&conn
->bytes_to_rcv
)) {
153 smc_rx_wait_data(smc
, &timeo
);
158 /* initialize variables for 1st iteration of subsequent loop */
159 /* could be just 1 byte, even after smc_rx_wait_data above */
160 readable
= atomic_read(&conn
->bytes_to_rcv
);
161 /* not more than what user space asked for */
162 copylen
= min_t(size_t, read_remaining
, readable
);
163 smc_curs_write(&cons
,
164 smc_curs_read(&conn
->local_tx_ctrl
.cons
, conn
),
166 /* determine chunks where to read from rcvbuf */
167 /* either unwrapped case, or 1st chunk of wrapped case */
168 chunk_len
= min_t(size_t,
169 copylen
, conn
->rmbe_size
- cons
.count
);
170 chunk_len_sum
= chunk_len
;
171 chunk_off
= cons
.count
;
172 for (chunk
= 0; chunk
< 2; chunk
++) {
173 if (!(flags
& MSG_TRUNC
)) {
174 rc
= memcpy_to_msg(msg
, rcvbuf_base
+ chunk_off
,
182 read_remaining
-= chunk_len
;
183 read_done
+= chunk_len
;
185 if (chunk_len_sum
== copylen
)
186 break; /* either on 1st or 2nd iteration */
187 /* prepare next (== 2nd) iteration */
188 chunk_len
= copylen
- chunk_len
; /* remainder */
189 chunk_len_sum
+= chunk_len
;
190 chunk_off
= 0; /* modulo offset in recv ring buffer */
194 if (!(flags
& MSG_PEEK
)) {
195 smc_curs_add(conn
->rmbe_size
, &cons
, copylen
);
196 /* increased in recv tasklet smc_cdc_msg_rcv() */
197 smp_mb__before_atomic();
198 atomic_sub(copylen
, &conn
->bytes_to_rcv
);
199 /* guarantee 0 <= bytes_to_rcv <= rmbe_size */
200 smp_mb__after_atomic();
201 smc_curs_write(&conn
->local_tx_ctrl
.cons
,
202 smc_curs_read(&cons
, conn
),
204 /* send consumer cursor update if required */
205 /* similar to advertising new TCP rcv_wnd if required */
206 smc_tx_consumer_update(conn
);
208 } while (read_remaining
);
213 /* Initialize receive properties on connection establishment. NB: not __init! */
214 void smc_rx_init(struct smc_sock
*smc
)
216 smc
->sk
.sk_data_ready
= smc_rx_data_ready
;