#include <net/inet_sock.h>
#include <net/tcp.h> /* For urgent data flags */
#include <rdma/ib_verbs.h>
+#include <linux/sched.h>
+
+#undef SDPSTATS_ON
+#undef SDP_PROFILING
+#undef CONFIG_INFINIBAND_SDP_DEBUG_DATA
+#undef CONFIG_INFINIBAND_SDP_DEBUG
-#undef SDP_LOCKS_CHECK
#define SDPSTATS_ON
#define SDP_PROFILING
+#define CONFIG_INFINIBAND_SDP_DEBUG_DATA
+#define CONFIG_INFINIBAND_SDP_DEBUG
#define _sdp_printk(func, line, level, sk, format, arg...) \
- printk(level "%s:%d sdp_sock(%5d %d:%d): " format, \
+ printk(level "%s:%d sdp_sock(%5d:%d %d:%d): " format, \
func, line, \
- current->pid, \
+ current->pid, smp_processor_id(), \
(sk) ? inet_sk(sk)->num : -1, \
(sk) ? ntohs(inet_sk(sk)->dport) : -1, ## arg)
#define sdp_printk(level, sk, format, arg...) \
sdp_printk(KERN_WARNING, sk, format , ## arg)
-#ifdef SDP_LOCKS_CHECK
-#define WARN_ON_UNLOCKED(sk, l) do {\
- if (unlikely(!spin_is_locked(l))) { \
- sdp_warn(sk, "lock " #l " should be locked\n"); \
- WARN_ON(1); \
- } \
-} while (0)
-
-#define WARN_ON_LOCKED(sk, l) do {\
- if (unlikely(spin_is_locked(l))) { \
- sdp_warn(sk, "lock " #l " should be unlocked\n"); \
- WARN_ON(1); \
- } \
-} while (0)
-#else
-#define WARN_ON_UNLOCKED(sk, l)
-#define WARN_ON_LOCKED(sk, l)
-#endif
-
#define rx_ring_lock(ssk, f) do { \
spin_lock_irqsave(&ssk->rx_ring.lock, f); \
} while (0)
struct sdpprf_log {
int idx;
int pid;
+ int cpu;
int sk_num;
int sk_dport;
struct sk_buff *skb;
l->pid = current->pid; \
l->sk_num = (sk) ? inet_sk(sk)->num : -1; \
l->sk_dport = (sk) ? ntohs(inet_sk(sk)->dport) : -1; \
+ l->cpu = smp_processor_id(); \
l->skb = s; \
snprintf(l->msg, sizeof(l->msg) - 1, format, ## arg); \
l->time = current_nsec(); \
struct list_head sock_list;
struct list_head accept_queue;
struct list_head backlog_queue;
- struct sk_buff_head rx_backlog;
+ struct sk_buff_head rx_ctl_q;
struct sock *parent;
struct work_struct rx_comp_work;
u16 urg_data;
u32 urg_seq;
u32 copied_seq;
- u32 rcv_nxt;
+#define rcv_nxt(ssk) atomic_read(&(ssk->rcv_nxt))
+ atomic_t rcv_nxt;
int write_seq;
int pushed_seq;
void sdp_rx_ring_init(struct sdp_sock *ssk);
int sdp_rx_ring_create(struct sdp_sock *ssk, struct ib_device *device);
void sdp_rx_ring_destroy(struct sdp_sock *ssk);
-int sdp_process_rx_q(struct sdp_sock *ssk);
int sdp_resize_buffers(struct sdp_sock *ssk, u32 new_size);
int sdp_init_buffers(struct sdp_sock *ssk, u32 new_size);
-void sdp_post_recvs(struct sdp_sock *ssk);
+void sdp_schedule_post_recvs(struct sdp_sock *ssk);
+void sdp_rx_comp_full(struct sdp_sock *ssk);
static inline void sdp_arm_rx_cq(struct sock *sk)
{
read_lock(&device_removal_lock);
- sdp_process_rx_q(ssk);
-
if (ssk->tx_ring.cq)
sdp_xmit_poll(ssk, 1);
* descriptor close, not protocol-sourced closes, because the
* reader process may not have drained the data yet!
*/
- while ((skb = __skb_dequeue(&sk->sk_receive_queue)) != NULL) {
+ while ((skb = skb_dequeue(&sk->sk_receive_queue)) != NULL) {
data_was_unread = 1;
__kfree_skb(skb);
}
TASK_INTERRUPTIBLE);
release_sock(sk);
if (list_empty(&ssk->accept_queue)) {
- sdp_dbg(sk, "%s schedule_timeout\n", __func__);
timeo = schedule_timeout(timeo);
- sdp_dbg(sk, "%s schedule_timeout done\n", __func__);
}
- sdp_dbg(sk, "%s lock_sock\n", __func__);
lock_sock(sk);
- sdp_dbg(sk, "%s lock_sock done\n", __func__);
err = 0;
if (!list_empty(&ssk->accept_queue))
break;
else if (sock_flag(sk, SOCK_URGINLINE) ||
!ssk->urg_data ||
before(ssk->urg_seq, ssk->copied_seq) ||
- !before(ssk->urg_seq, ssk->rcv_nxt)) {
- answ = ssk->rcv_nxt - ssk->copied_seq;
+ !before(ssk->urg_seq, rcv_nxt(ssk))) {
+ answ = rcv_nxt(ssk) - ssk->copied_seq;
/* Subtract 1, if FIN is in queue. */
if (answ && !skb_queue_empty(&sk->sk_receive_queue))
sk->sk_route_caps |= NETIF_F_SG | NETIF_F_NO_CSUM;
- skb_queue_head_init(&ssk->rx_backlog);
+ skb_queue_head_init(&ssk->rx_ctl_q);
atomic_set(&ssk->mseq_ack, 0);
{
int i;
for (i = 0; i < recv_poll; ++i) {
- if (!sdp_process_rx_q(sdp_sk(sk))) {
+ if (!skb_queue_empty(&sk->sk_receive_queue)) {
++recv_poll_hit;
return 0;
}
return -EAGAIN;
}
-static void sdp_rcv_space_adjust(struct sock *sk)
-{
- sdp_post_recvs(sdp_sk(sk));
- sdp_post_sends(sdp_sk(sk), 0);
-}
-
static unsigned int sdp_current_mss(struct sock *sk, int large_allowed)
{
/* TODO */
lock_sock(sk);
sdp_dbg_data(sk, "%s\n", __func__);
-// sdp_prf(sk, skb, "Read from user");
+ sdp_prf(sk, skb, "Read from user");
err = -ENOTCONN;
if (sk->sk_state == TCP_LISTEN)
if (!skb)
break;
- if ((skb_transport_header(skb))[0] == SDP_MID_DISCONN)
- goto found_fin_ok;
+ BUG_ON((skb_transport_header(skb))[0] != SDP_MID_DATA);
if (before(*seq, TCP_SKB_CB(skb)->seq)) {
- printk(KERN_INFO "recvmsg bug: copied %X "
- "seq %X\n", *seq, TCP_SKB_CB(skb)->seq);
+ sdp_warn(sk, "recvmsg bug: copied %X seq %X\n",
+ *seq, TCP_SKB_CB(skb)->seq);
+ sdp_reset(sk);
break;
}
sdp_dbg_data(sk, "%s: done copied %d target %d\n", __func__, copied, target);
- sdp_rcv_space_adjust(sk);
+ sdp_schedule_post_recvs(sdp_sk(sk));
skip_copy:
if (ssk->urg_data && after(ssk->copied_seq, ssk->urg_seq))
ssk->urg_data = 0;
continue;
offset = 0;
- if (!(flags & MSG_PEEK))
- sk_eat_skb(sk, skb, 0);
-
- continue;
-found_fin_ok:
- ++*seq;
- if (!(flags & MSG_PEEK))
- sk_eat_skb(sk, skb, 0);
-
- break;
+ if (!(flags & MSG_PEEK)) {
+ skb_unlink(skb, &sk->sk_receive_queue);
+ __kfree_skb(skb);
+ }
} while (len > 0);
- release_sock(sk);
- return copied;
-
+ err = copied;
out:
release_sock(sk);
return err;
}
}
-/* lock_sock must be taken before calling this - since rx_ring.head is not
- * protected (although being atomic
- */
static int sdp_post_recv(struct sdp_sock *ssk)
{
struct sdp_buf *rx_req;
gfp_t gfp_page;
int ret = 0;
- WARN_ON_UNLOCKED(&ssk->isk.sk, &ssk->rx_ring.lock);
/* Now, allocate and repost recv */
/* TODO: allocate from cache */
gfp_page = GFP_HIGHUSER;
}
+ sdp_prf(&ssk->isk.sk, skb, "Posting skb");
/* FIXME */
BUG_ON(!skb);
h = (struct sdp_bsdh *)skb->head;
rx_wr.sg_list = ibsge;
rx_wr.num_sge = frags + 1;
rc = ib_post_recv(ssk->qp, &rx_wr, &bad_wr);
- SDPSTATS_COUNTER_INC(post_recv);
atomic_inc(&ssk->rx_ring.head);
if (unlikely(rc)) {
sdp_warn(&ssk->isk.sk, "ib_post_recv failed with status %d\n", rc);
+
+ lock_sock(&ssk->isk.sk);
sdp_reset(&ssk->isk.sk);
+ release_sock(&ssk->isk.sk);
+
ret = -1;
}
+ SDPSTATS_COUNTER_INC(post_recv);
atomic_add(SDP_MAX_SEND_SKB_FRAGS, &sdp_current_mem_usage);
return ret;
}
-/* lock_sock must be taken before calling this */
-static void _sdp_post_recvs(struct sdp_sock *ssk)
+static inline int sdp_post_recvs_needed(struct sdp_sock *ssk)
{
struct sock *sk = &ssk->isk.sk;
int scale = ssk->rcvbuf_scale;
-
- WARN_ON_UNLOCKED(&ssk->isk.sk, &ssk->rx_ring.lock);
-
- if (unlikely(!ssk->id || ((1 << sk->sk_state) &
- (TCPF_CLOSE | TCPF_TIME_WAIT)))) {
- return;
- }
+ int buffer_size = SDP_HEAD_SIZE + ssk->recv_frags * PAGE_SIZE;
+ unsigned long max_bytes;
if (top_mem_usage &&
(top_mem_usage * 0x100000) < atomic_read(&sdp_current_mem_usage) * PAGE_SIZE)
scale = 1;
- while ((likely(ring_posted(ssk->rx_ring) < SDP_RX_SIZE) &&
- (ring_posted(ssk->rx_ring) - SDP_MIN_TX_CREDITS) *
- (SDP_HEAD_SIZE + ssk->recv_frags * PAGE_SIZE) +
- ssk->rcv_nxt - ssk->copied_seq < sk->sk_rcvbuf * scale) ||
- unlikely(ring_posted(ssk->rx_ring) < SDP_MIN_TX_CREDITS)) {
- if (sdp_post_recv(ssk))
- break;
+ max_bytes = sk->sk_rcvbuf * scale;
+
+ if (unlikely(ring_posted(ssk->rx_ring) >= SDP_RX_SIZE)) {
+ sdp_prf(sk, NULL, "rx ring is full");
+ return 0;
}
+
+ if (likely(ring_posted(ssk->rx_ring) > SDP_MIN_TX_CREDITS)) {
+ unsigned long bytes_in_process =
+ (ring_posted(ssk->rx_ring) - SDP_MIN_TX_CREDITS) * buffer_size;
+ bytes_in_process += rcv_nxt(ssk) - ssk->copied_seq;
+
+ if (bytes_in_process >= max_bytes) {
+ sdp_prf(sk, NULL, "bytes_in_process:%ld > max_bytes:%ld",
+ bytes_in_process, max_bytes);
+ return 0;
+ }
+ }
+
+ return 1;
}
-void sdp_post_recvs(struct sdp_sock *ssk)
+static inline void sdp_post_recvs(struct sdp_sock *ssk)
{
- unsigned long flags;
+ int rc = 0;
+again:
+ ssk->posts_in_process = 1;
- rx_ring_lock(ssk, flags);
- _sdp_post_recvs(ssk);
- rx_ring_unlock(ssk, flags);
+ do {
+ if (!sdp_post_recvs_needed(ssk))
+ break;
+
+ rc = sdp_post_recv(ssk);
+ } while (!rc);
+
+ sk_stream_mem_reclaim(&ssk->isk.sk);
+
+ ssk->posts_in_process = 0;
+ if (sdp_post_recvs_needed(ssk))
+ goto again;
+}
+
+void sdp_schedule_post_recvs(struct sdp_sock *ssk)
+{
+ struct sock *sk = &ssk->isk.sk;
+
+ WARN_ON_LOCKED(&ssk->isk.sk, &ssk->rx_ring.lock);
+
+ if (unlikely(!ssk->id || ((1 << sk->sk_state) &
+ (TCPF_CLOSE | TCPF_TIME_WAIT)))) {
+ sdp_prf(sk, NULL, "socket is closed - not posting");
+ return;
+ }
+
+ if (!ssk->posts_in_process && sdp_post_recvs_needed(ssk))
+ queue_work(rx_comp_wq, &ssk->rx_comp_work);
}
static inline struct sk_buff *sdp_sock_queue_rcv_skb(struct sock *sk,
{
int skb_len;
struct sdp_sock *ssk = sdp_sk(sk);
- struct sk_buff *tail = NULL;
/* not needed since sk_rmem_alloc is not currently used
* TODO - remove this?
skb_len = skb->len;
- TCP_SKB_CB(skb)->seq = ssk->rcv_nxt;
- ssk->rcv_nxt += skb_len;
+ TCP_SKB_CB(skb)->seq = rcv_nxt(ssk);
+ atomic_add(skb_len, &ssk->rcv_nxt);
- if (likely(skb_len && (tail = skb_peek_tail(&sk->sk_receive_queue))) &&
- unlikely(skb_tailroom(tail) >= skb_len)) {
- skb_copy_bits(skb, 0, skb_put(tail, skb_len), skb_len);
- __kfree_skb(skb);
- skb = tail;
- } else
- skb_queue_tail(&sk->sk_receive_queue, skb);
+ skb_queue_tail(&sk->sk_receive_queue, skb);
if (!sock_flag(sk, SOCK_DEAD))
sk->sk_data_ready(sk, skb_len);
return skb;
}
-/* this must be called while sock_lock is taken */
-static int sdp_process_rx_skb(struct sdp_sock *ssk, struct sk_buff *skb)
+/* socket lock should be taken before calling this */
+static int sdp_process_rx_ctl_skb(struct sdp_sock *ssk, struct sk_buff *skb)
{
+ struct sdp_bsdh *h = (struct sdp_bsdh *)skb->data;
struct sock *sk = &ssk->isk.sk;
- int frags;
- struct sdp_bsdh *h;
- int pagesz, i;
-
- h = (struct sdp_bsdh *)skb->data;
-
- frags = skb_shinfo(skb)->nr_frags;
- pagesz = PAGE_ALIGN(skb->data_len);
- skb_shinfo(skb)->nr_frags = pagesz / PAGE_SIZE;
-
- for (i = skb_shinfo(skb)->nr_frags;
- i < frags; ++i) {
- put_page(skb_shinfo(skb)->frags[i].page);
- skb->truesize -= PAGE_SIZE;
- }
-
- if (unlikely(h->flags & SDP_OOB_PEND))
- sk_send_sigurg(sk);
-
- skb_pull(skb, sizeof(struct sdp_bsdh));
switch (h->mid) {
case SDP_MID_DATA:
- if (unlikely(skb->len <= 0)) {
- __kfree_skb(skb);
- break;
- }
+ WARN_ON(!(sk->sk_shutdown & RCV_SHUTDOWN));
- if (unlikely(sk->sk_shutdown & RCV_SHUTDOWN)) {
- /* got data in RCV_SHUTDOWN */
- if (sk->sk_state == TCP_FIN_WAIT1) {
- /* go into abortive close */
- sdp_exch_state(sk, TCPF_FIN_WAIT1,
- TCP_TIME_WAIT);
+ sdp_warn(sk, "DATA after socket rcv was shutdown\n");
- sk->sk_prot->disconnect(sk, 0);
- }
+ /* got data in RCV_SHUTDOWN */
+ if (sk->sk_state == TCP_FIN_WAIT1) {
+ sdp_warn(sk, "socket in shutdown, state = FIN_WAIT1 and got data\n");
+ /* go into abortive close */
+ sdp_exch_state(sk, TCPF_FIN_WAIT1,
+ TCP_TIME_WAIT);
- __kfree_skb(skb);
- break;
+ sk->sk_prot->disconnect(sk, 0);
}
- skb = sdp_sock_queue_rcv_skb(sk, skb);
- if (unlikely(h->flags & SDP_OOB_PRES))
- sdp_urg(ssk, skb);
+
break;
case SDP_MID_DISCONN:
- __kfree_skb(skb);
+ sdp_dbg_data(sk, "Handling RX disconnect\n");
sdp_fin(sk);
break;
case SDP_MID_CHRCVBUF:
+ sdp_dbg_data(sk, "Handling RX CHRCVBUF\n");
sdp_handle_resize_request(ssk,
(struct sdp_chrecvbuf *)skb->data);
- __kfree_skb(skb);
break;
case SDP_MID_CHRCVBUF_ACK:
+ sdp_dbg_data(sk, "Handling RX CHRCVBUF_ACK\n");
sdp_handle_resize_ack(ssk, (struct sdp_chrecvbuf *)skb->data);
- __kfree_skb(skb);
break;
default:
/* TODO: Handle other messages */
- printk(KERN_WARNING "SDP: FIXME MID %d\n", h->mid);
+ sdp_warn(sk, "SDP: FIXME MID %d\n", h->mid);
+ }
+
+ __kfree_skb(skb);
+
+ return 0;
+}
+
+static int sdp_process_rx_skb(struct sdp_sock *ssk, struct sk_buff *skb)
+{
+ struct sock *sk = &ssk->isk.sk;
+ int frags;
+ struct sdp_bsdh *h;
+ int pagesz, i;
+ unsigned long mseq_ack;
+ int credits_before;
+
+ h = (struct sdp_bsdh *)skb->data;
+
+ SDPSTATS_HIST_LINEAR(credits_before_update, tx_credits(ssk));
+
+ mseq_ack = ntohl(h->mseq_ack);
+ credits_before = tx_credits(ssk);
+ atomic_set(&ssk->tx_ring.credits, mseq_ack - ring_head(ssk->tx_ring) + 1 +
+ ntohs(h->bufs));
+ if (mseq_ack >= ssk->nagle_last_unacked)
+ ssk->nagle_last_unacked = 0;
+
+ sdp_prf(&ssk->isk.sk, skb, "RX %s +%d c:%d->%d mseq:%d ack:%d",
+ mid2str(h->mid), ntohs(h->bufs), credits_before,
+ tx_credits(ssk), ntohl(h->mseq), ntohl(h->mseq_ack));
+
+ frags = skb_shinfo(skb)->nr_frags;
+ pagesz = PAGE_ALIGN(skb->data_len);
+ skb_shinfo(skb)->nr_frags = pagesz / PAGE_SIZE;
+
+ for (i = skb_shinfo(skb)->nr_frags; i < frags; ++i) {
+ put_page(skb_shinfo(skb)->frags[i].page);
+ skb->truesize -= PAGE_SIZE;
+ }
+
+/* if (unlikely(h->flags & SDP_OOB_PEND))
+ sk_send_sigurg(sk);*/
+
+ if (h->mid != SDP_MID_DATA || unlikely(sk->sk_shutdown & RCV_SHUTDOWN)) {
+ sdp_warn(sk, "Control skb - queing to control queue\n");
+ skb_queue_tail(&ssk->rx_ctl_q, skb);
+ return 0;
+ }
+
+ skb_pull(skb, sizeof(struct sdp_bsdh));
+
+ if (unlikely(skb->len <= 0)) {
__kfree_skb(skb);
}
struct sk_buff *skb;
struct sdp_bsdh *h;
struct sock *sk = &ssk->isk.sk;
- int credits_before;
- unsigned long mseq_ack;
+ int mseq;
skb = sdp_recv_completion(ssk, wc->wr_id);
if (unlikely(!skb))
h = (struct sdp_bsdh *)skb->data;
SDP_DUMP_PACKET(&ssk->isk.sk, "RX", skb, h);
skb_reset_transport_header(skb);
- atomic_set(&ssk->mseq_ack, ntohl(h->mseq));
- if (mseq_ack(ssk) != (int)wc->wr_id)
- printk(KERN_WARNING "SDP BUG! mseq %d != wrid %d\n",
- mseq_ack(ssk), (int)wc->wr_id);
- SDPSTATS_HIST_LINEAR(credits_before_update, tx_credits(ssk));
-
- mseq_ack = ntohl(h->mseq_ack);
- credits_before = tx_credits(ssk);
- atomic_set(&ssk->tx_ring.credits, mseq_ack - ring_head(ssk->tx_ring) + 1 +
- ntohs(h->bufs));
- if (mseq_ack >= ssk->nagle_last_unacked)
- ssk->nagle_last_unacked = 0;
-
- sdp_prf(&ssk->isk.sk, skb, "RX %s bufs=%d c before:%d after:%d "
- "mseq:%d, ack:%d", mid2str(h->mid), ntohs(h->bufs), credits_before,
- tx_credits(ssk), ntohl(h->mseq), ntohl(h->mseq_ack));
+ mseq = ntohl(h->mseq);
+ atomic_set(&ssk->mseq_ack, mseq);
+ if (mseq != (int)wc->wr_id)
+ sdp_warn(sk, "SDP BUG! mseq %d != wrid %d\n",
+ mseq, (int)wc->wr_id);
return skb;
}
}
/* only from interrupt.
- * drain rx cq into rx_backlog queue */
static int sdp_poll_rx_cq(struct sdp_sock *ssk)
{
struct ib_cq *cq = ssk->rx_ring.cq;
skb = sdp_process_rx_wc(ssk, wc);
if (!skb)
continue;
- skb_queue_tail(&ssk->rx_backlog, skb);
+
+ sdp_process_rx_skb(ssk, skb);
wc_processed++;
}
} while (n == SDP_NUM_WC);
return wc_processed;
}
-int sdp_process_rx_q(struct sdp_sock *ssk)
+void sdp_rx_comp_work(struct work_struct *work)
{
- struct sk_buff *skb;
+ struct sdp_sock *ssk = container_of(work, struct sdp_sock, rx_comp_work);
struct sock *sk = &ssk->isk.sk;
- unsigned long flags;
+ int xmit_poll_force;
+ struct sk_buff *skb;
- if (!ssk->rx_backlog.next || !ssk->rx_backlog.prev) {
- sdp_warn(&ssk->isk.sk, "polling a zeroed rx_backlog!!!! %p\n", &ssk->rx_backlog);
- return 0;
- }
+ sdp_prf(sk, NULL, "%s", __func__);
+
+ if (unlikely(!ssk->rx_ring.cq))
+ return;
- if (skb_queue_empty(&ssk->rx_backlog)) {
- SDPSTATS_COUNTER_INC(rx_poll_miss);
- return -EAGAIN;
+ if (unlikely(!ssk->poll_cq)) {
+ struct rdma_cm_id *id = ssk->id;
+ sdp_warn(sk, "poll cq is 0. socket was reset or wasn't initialized\n");
+ if (id && id->qp)
+ rdma_notify(id, RDMA_CM_EVENT_ESTABLISHED);
+ return;
}
- /* update credits */
- sdp_post_sends(ssk, 0);
+ sdp_post_recvs(ssk);
- spin_lock_irqsave(&ssk->rx_backlog.lock, flags);
- while ((skb = __skb_dequeue(&ssk->rx_backlog))) {
- sdp_process_rx_skb(ssk, skb);
+ if (!(sk->sk_write_pending && (tx_credits(ssk) > SDP_MIN_TX_CREDITS)) &&
+ !credit_update_needed(ssk) &&
+ skb_queue_empty(&ssk->rx_ctl_q)) {
+ sdp_prf(&ssk->isk.sk, NULL, "only post recvs");
+ return;
}
- spin_unlock_irqrestore(&ssk->rx_backlog.lock, flags);
- if (sk->sk_sleep && waitqueue_active(sk->sk_sleep))
- sk_stream_write_space(&ssk->isk.sk);
+ lock_sock(sk);
- return 0;
-}
+ sdp_post_sends(ssk, 0);
-static void sdp_rx_comp_work(struct work_struct *work)
-{
- struct sdp_sock *ssk = container_of(work, struct sdp_sock, rx_comp_work);
- struct sock *sk = &ssk->isk.sk;
- struct ib_cq *rx_cq;
+ while ((skb = skb_dequeue(&ssk->rx_ctl_q))) {
+ sdp_process_rx_ctl_skb(ssk, skb);
+ }
- lock_sock(sk);
- rx_cq = ssk->rx_ring.cq;
- if (unlikely(!rx_cq))
- goto out;
+ sk_stream_mem_reclaim(sk);
- if (unlikely(!ssk->poll_cq)) {
- struct rdma_cm_id *id = ssk->id;
- sdp_warn(sk, "poll cq is 0. socket was reset or wasn't initialized\n");
- if (id && id->qp)
- rdma_notify(id, RDMA_CM_EVENT_ESTABLISHED);
- goto out;
- }
+ xmit_poll_force = sk->sk_write_pending && (tx_credits(ssk) > SDP_MIN_TX_CREDITS);
- sdp_process_rx_q(ssk);
- sdp_xmit_poll(ssk, 1); /* if has pending tx because run out of tx_credits - xmit it */
- release_sock(sk);
- sk_mem_reclaim(sk);
- lock_sock(sk);
- rx_cq = ssk->rx_ring.cq;
- if (unlikely(!rx_cq))
- goto out;
-
- sdp_process_rx_q(ssk);
- sdp_xmit_poll(ssk, 1);
+ if (credit_update_needed(ssk) || xmit_poll_force) {
+ /* if has pending tx because run out of tx_credits - xmit it */
+ sdp_prf(sk, NULL, "Processing to free pending sends");
+ sdp_xmit_poll(ssk, xmit_poll_force);
+ sdp_prf(sk, NULL, "Sending credit update if needed");
+ sdp_post_sends(ssk, 0);
+ }
-out:
release_sock(sk);
}
struct sdp_sock *ssk = sdp_sk(sk);
unsigned long flags;
int wc_processed = 0;
+ int credits_before;
sdp_dbg_data(&ssk->isk.sk, "rx irq called\n");
rx_ring_lock(ssk, flags);
+ credits_before = tx_credits(ssk);
+
if (unlikely(!ssk->poll_cq))
sdp_warn(sk, "poll cq is 0. socket was reset or wasn't initialized\n");
sdp_prf(&ssk->isk.sk, NULL, "processed %d", wc_processed);
if (wc_processed) {
- _sdp_post_recvs(ssk);
+ sdp_prf(&ssk->isk.sk, NULL, "credits: %d -> %d",
+ credits_before, tx_credits(ssk));
- /* Best was to send credit update from here */
-/* sdp_post_credits(ssk); */
-
- /* issue sdp_rx_comp_work() */
queue_work(rx_comp_wq, &ssk->rx_comp_work);
}
-
sdp_arm_rx_cq(sk);
out:
}
rx_cq = ib_create_cq(device, sdp_rx_irq, sdp_rx_cq_event_handler,
- &ssk->isk.sk, SDP_RX_SIZE, 0);
+ &ssk->isk.sk, SDP_RX_SIZE, IB_CQ_VECTOR_LEAST_ATTACHED);
if (IS_ERR(rx_cq)) {
rc = PTR_ERR(rx_cq);
projects / users / jedix / linux-maple.git / commitdiff