<commit_before>/*
 * Open-MX
 * Copyright © INRIA 2007-2008 (see AUTHORS file)
 *
 * The development of this software has been funded by Myricom, Inc.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or (at
 * your option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 *
 * See the GNU General Public License in COPYING.GPL for more details.
 */

#include <linux/kernel.h>
#include <linux/slab.h>
#include <linux/spinlock.h>
#include <linux/idr.h>
#include <linux/kref.h>
#include <linux/timer.h>
#include <linux/workqueue.h>

#include "omx_misc.h"
#include "omx_hal.h"
#include "omx_wire_access.h"
#include "omx_common.h"
#include "omx_iface.h"
#include "omx_peer.h"
#include "omx_endpoint.h"
#include "omx_reg.h"
#include "omx_dma.h"
#ifndef OMX_DISABLE_SHARED
#include "omx_shared.h"
#endif

/**************************
 * Pull-specific Constants
 */

#define OMX_PULL_BLOCK_LENGTH_MAX (OMX_PULL_REPLY_LENGTH_MAX*OMX_PULL_REPLY_PER_BLOCK)

#define OMX_PULL_RETRANSMIT_TIMEOUT_MS	1000
#define OMX_PULL_RETRANSMIT_TIMEOUT_JIFFIES (OMX_PULL_RETRANSMIT_TIMEOUT_MS*HZ/1000)

#define OMX_PULL_BLOCK_DESCS_NR 2

#ifdef OMX_MX_WIRE_COMPAT
#if OMX_PULL_REPLY_LENGTH_MAX >= 65536
#error Cannot store rdma offsets > 65535 in 16bits offsets on the wire
#endif
#endif

/**********************
 * Pull-specific Types
 */

#if OMX_PULL_REPLY_PER_BLOCK & (OMX_PULL_REPLY_PER_BLOCK-1)
/* we don't want to divide by a non-power-of-two */
#error Need a power of two as the number of replies per pull block
#elif OMX_PULL_REPLY_PER_BLOCK > 64
#error Cannot request more than 64 replies per pull block
#elif OMX_PULL_REPLY_PER_BLOCK > 32
typedef uint64_t omx_block_frame_bitmask_t;
#elif OMX_PULL_REPLY_PER_BLOCK > 16
typedef uint32_t omx_block_frame_bitmask_t;
#elif OMX_PULL_REPLY_PER_BLOCK > 8
typedef uint16_t omx_block_frame_bitmask_t;
#else
typedef uint8_t omx_block_frame_bitmask_t;
#endif

enum omx_pull_handle_status {
	/*
	 * The handle is normal, being processed as usual and its timeout handler is running ok.
	 * It is queued on the endpoint running_list.
	 */
	OMX_PULL_HANDLE_STATUS_OK,

	/*
	 * The handle has been removed from the idr, but the timeout handler is still running.
	 * It is queued on the endpoint done_but_timer_list.
	 * Either the pull has completed (or aborted on error), or the endpoint is being closed and
	 * all handles have been scheduled for removal.
	 * The timeout handler must exit next time it runs. It will release the reference on the handle,
	 * dequeue it, and the handle may be destroyed.
	 */
	OMX_PULL_HANDLE_STATUS_TIMER_MUST_EXIT,

	/*
	 * The handle has been removed from the idr and the endpoint lists, and its timeout handler has exited
	 * and released its reference.
	 * Either the pull has completed (or aborted on error) and the handler caught the MUST_EXIT status,
	 * or the timeout was reached and its handler aborted the handle directly.
	 */
	OMX_PULL_HANDLE_STATUS_TIMER_EXITED,
};

struct omx_pull_block_desc {
	uint32_t frame_index;
	uint32_t block_length;
	uint32_t first_frame_offset;
	omx_block_frame_bitmask_t frames_missing_bitmap; /* frames not received at all */
};

struct omx_pull_handle {
	struct kref refcount;
	struct list_head list_elt; /* always queued on one of the endpoint lists */
	int idr_index; /* that's what idr_get_new wants */
	uint32_t magic; /* 8bits endpoint index + 24bits handle generation number */

	/* timer for retransmission */
	struct timer_list retransmit_timer;
	uint64_t last_retransmit_jiffies;

	/* global pull fields */
	struct omx_endpoint * endpoint;
	struct omx_user_region * region;
	uint32_t total_length;
	uint32_t puller_rdma_offset;
	uint32_t pulled_rdma_offset;

	/* current status */
	spinlock_t lock;
	enum omx_pull_handle_status status;
	uint32_t remaining_length;
	uint32_t frame_index; /* index of the first requested frame */
	uint32_t next_frame_index; /* index of the frame to request */
	uint32_t nr_requested_frames; /* number of frames requested */
	uint32_t nr_missing_frames; /* frames requested but not received yet */
	uint32_t nr_copying_frames; /* frames received but not copied yet*/
	uint32_t nr_valid_block_descs;
	uint32_t already_rerequested_blocks; /* amount of first blocks that were requested again since the last timer */
	struct omx_pull_block_desc block_desc[OMX_PULL_BLOCK_DESCS_NR];

#ifdef CONFIG_NET_DMA
	struct dma_chan *dma_chan;
	dma_cookie_t dma_last_cookie;
	struct sk_buff_head dma_skb_queue;
	struct work_struct deferred_dma_wait_work;
#endif

	/* completion event */
	struct omx_evt_pull_done done_event;

	/* pull packet header */
	struct omx_hdr pkt_hdr;
};

static void omx_pull_handle_timeout_handler(unsigned long data);

#ifdef CONFIG_NET_DMA
static void omx_pull_handle_poll_dma_completions(struct omx_pull_handle *handle);
static int omx_pull_handle_deferred_wait_dma_completions(struct omx_pull_handle *handle);
#else
#define omx_pull_handle_poll_dma_completions(ph) /* nothing */
#define omx_pull_handle_deferred_wait_dma_completions(ph) 0 /* always completed */
#endif

/*
 * Notes about locking:
 *
 * Each handle owns a spin_lock that protects the actual pull status (frame index, ...).
 * It also protects its handle status and its queueing in the endpoint lists and idr.
 * This lock is always taken *before* the endpoint pull handle lock.
 *
 * The handle does not own a reference on the endpoint. It is always queued in one
 * endpoint lists, and the endpoint closing will enforce its detruction.
 * When the endpoint starts to be closed, it calls prepare_exit which sets all handle
 * to timer_must_exit but cannot wait for them because of the possible interrupt
 * context. Later, the cleanup thread will cleanup the endpoint, including destroying
 * the handle timers that are still running.
 *
 * The pile of handles for an endpoint is protected by a rwlock. It is taken for
 * reading when acquiring an handle (when a pull reply or nack mcp arrives, likely
 * in a bottom half). It is taken for writing when creating a handle (when the
 * application request a pull), finishing a handle (when a pull reply arrives
 * and completes the pull request, likely in a bottom half), and when destroying
 * and remaining handles (when an endpoint is closed).
 * Since a bottom half and the application may both acquire the rwlock for
 * writing, we must always disable bottom halves when taking the rwlock for
 * either read or writing.
 */

/*
 * Notes about retransmission:
 *
 * The puller request 2 blocks of data, and waits for OMX_PULL_REPLY_PER_BLOCK
 * replies for each of them.
 *
 * A timer is set to detect when nothing has been received for a while.  It is
 * updated every time a new reply is received. This timer repost requests to
 * get current blocks (using descriptors that were cached in the pull handle).
 *
 * Additionally, if the second block completes before the first one, there is
 * a good chance that one packet got lost for the first block. In this case,
 * we optimistically re-request the first block.
 * To avoid re-requesting too often, we do it only once per timeout.
 *
 * In the end, the timer is only called if:
 * + one packet is lost for both the first and the second current block
 * + or one packet is missing in the first block after one optimistic re-request.
 * So the timeout doesn't need to be short, 1 second is enough.
 */

#ifdef OMX_DRIVER_DEBUG
/* defined as module parameters */
extern unsigned long omx_PULL_REQ_packet_loss;
extern unsigned long omx_PULL_REPLY_packet_loss;
/* index between 0 and the above limit */
static unsigned long omx_PULL_REQ_packet_loss_index = 0;
static unsigned long omx_PULL_REPLY_packet_loss_index = 0;
#endif /* OMX_DRIVER_DEBUG */

/********************************
 * Pull handle bitmap management
 */

static INLINE void
omx_pull_handle_append_needed_frames(struct omx_pull_handle * handle,
				     uint32_t block_length,
				     uint32_t first_frame_offset)
{
	struct omx_pull_block_desc *desc;
	omx_block_frame_bitmask_t new_mask;
	int new_frames;

	new_frames = (first_frame_offset + block_length
		      + OMX_PULL_REPLY_LENGTH_MAX-1) / OMX_PULL_REPLY_LENGTH_MAX;
	new_mask = ((omx_block_frame_bitmask_t) -1) >> (OMX_PULL_REPLY_PER_BLOCK-new_frames);

	desc = &handle->block_desc[handle->nr_valid_block_descs];
	desc->frame_index = handle->next_frame_index;
	desc->block_length = block_length;
	desc->first_frame_offset = first_frame_offset;
	desc->frames_missing_bitmap = new_mask;

	handle->nr_requested_frames += new_frames;
	handle->nr_missing_frames += new_frames;
	handle->next_frame_index += new_frames;
	handle->remaining_length -= block_length;
	handle->nr_valid_block_descs++;

	dprintk(PULL, "appending block #%d with %d new frames to pull handle %p, now requested %ld-%ld\n",
		handle->nr_valid_block_descs-1, new_frames, handle,
		(unsigned long) handle->frame_index, (unsigned long) handle->next_frame_index-1);
}

static INLINE void
omx_pull_handle_first_block_done(struct omx_pull_handle * handle)
{
	uint32_t first_block_frames = min(handle->nr_requested_frames, (uint32_t) OMX_PULL_REPLY_PER_BLOCK);

	handle->frame_index += first_block_frames;
	handle->nr_requested_frames -= first_block_frames;
	handle->nr_valid_block_descs--;
	if (handle->already_rerequested_blocks)
		handle->already_rerequested_blocks--;
	memmove(&handle->block_desc[0], &handle->block_desc[1],
		sizeof(struct omx_pull_block_desc) * handle->nr_valid_block_descs);
	handle->block_desc[OMX_PULL_BLOCK_DESCS_NR-1].frames_missing_bitmap = 0; /* make sure the invalid block descs are easy to check */

	dprintk(PULL, "first block of pull handle %p done, removing %d requested frames, now requested %ld-%ld\n",
		handle, first_block_frames,
		(unsigned long) handle->frame_index, (unsigned long) handle->next_frame_index-1);
}

/**********************************
 * Pull handle acquiring/releasing
 */

/*
 * Acquire a handle.
 *
 * Either another reference on this handle should be owned,
 * or the endpoint lock should be hold.
 */
static INLINE void
omx_pull_handle_acquire(struct omx_pull_handle * handle)
{
	kref_get(&handle->refcount);
}

/*
 * Actual freeing of a handle when the last reference is released
 */
static void
__omx_pull_handle_last_release(struct kref * kref)
{
	struct omx_pull_handle * handle = container_of(kref, struct omx_pull_handle, refcount);

	dprintk(KREF, "releasing the last reference on pull handle %p\n",
		handle);

	BUG_ON(handle->status != OMX_PULL_HANDLE_STATUS_TIMER_EXITED);

	/* release the region now that we are sure that nobody else uses it */
	omx_user_region_release(handle->region);

	kfree(handle);
}

/*
 * Release an acquired pull handle
 */
static INLINE void
omx_pull_handle_release(struct omx_pull_handle * handle)
{
	kref_put(&handle->refcount, __omx_pull_handle_last_release);
}


/***************************************
 * Per-endpoint pull handles management
 */

int
omx_endpoint_pull_handles_init(struct omx_endpoint * endpoint)
{
	INIT_LIST_HEAD(&endpoint->pull_handles_running_list);
	INIT_LIST_HEAD(&endpoint->pull_handles_done_but_timer_list);
	idr_init(&endpoint->pull_handles_idr);
	rwlock_init(&endpoint->pull_handles_lock);
	return 0;
}

/*
 * Called when the last reference on the endpoint is removed,
 * possibly from unsafe context, cannot del_timer_sync() then.
 */
void
omx_endpoint_pull_handles_prepare_exit(struct omx_endpoint * endpoint)
{
	/*
	 * ask all pull handles of the endpoint to stop their timer.
	 * but we can't take endpoint->pull_handles_lock before handle->lock since that would deadlock
	 * so we use a tricky loop to take locks in order
	 */

	write_lock_bh(&endpoint->pull_handles_lock);
	while (!list_empty(&endpoint->pull_handles_running_list)) {
		struct omx_pull_handle * handle;

		/* get the first handle of the list, acquire it and release the list lock */
		handle = list_first_entry(&endpoint->pull_handles_running_list, struct omx_pull_handle, list_elt);
		omx_pull_handle_acquire(handle);
		write_unlock_bh(&endpoint->pull_handles_lock);

		/* take the handle lock and check the status in case it changed while the lock was released */
		spin_lock_bh(&handle->lock);
		if (handle->status == OMX_PULL_HANDLE_STATUS_OK) {
			/* the handle didn't change, do our stuff */
			handle->status = OMX_PULL_HANDLE_STATUS_TIMER_MUST_EXIT;

			/*
			 * remove from the idr so that no incoming packet can find it anymore,
			 * and move to the done_but_timer list
			 */
			dprintk(PULL, "moving handle %p to the done_but_timer list and removing from idr\n", handle);
			write_lock(&endpoint->pull_handles_lock);
			idr_remove(&endpoint->pull_handles_idr, handle->idr_index);
			list_move(&handle->list_elt, &endpoint->pull_handles_done_but_timer_list);
			write_unlock(&endpoint->pull_handles_lock);
		} else {
			/* the handle has been moved out of the list, it means that the timer is done, nothing to done */
		}
		spin_unlock_bh(&handle->lock);
		omx_pull_handle_release(handle);

		write_lock_bh(&endpoint->pull_handles_lock);
	}
	write_unlock_bh(&endpoint->pull_handles_lock);

	idr_destroy(&endpoint->pull_handles_idr);
}

/*
 * Called when cleaning the endpoint, always from the cleanup thread, may del_timer_sync() then.
 */
void
omx_endpoint_pull_handles_force_exit(struct omx_endpoint * endpoint)
{
	might_sleep();

	write_lock_bh(&endpoint->pull_handles_lock);
	while (!list_empty(&endpoint->pull_handles_done_but_timer_list)) {
		struct omx_pull_handle * handle;
		int ret;

		/* get the first handle of the list, acquire it and release the list lock */
		handle = list_first_entry(&endpoint->pull_handles_done_but_timer_list, struct omx_pull_handle, list_elt);
		omx_pull_handle_acquire(handle);
		write_unlock_bh(&endpoint->pull_handles_lock);

		dprintk(PULL, "stopping handle %p timer with del_sync_timer\n", handle);
		ret = del_timer_sync(&handle->retransmit_timer);
		spin_lock_bh(&handle->lock);
		if (ret) {
			dprintk(PULL, "del_timer_sync stopped pull handle %p timer\n", handle);

			/* we deactivated the timer, cleanup ourself */
			BUG_ON(handle->status != OMX_PULL_HANDLE_STATUS_TIMER_MUST_EXIT);
			handle->status = OMX_PULL_HANDLE_STATUS_TIMER_EXITED;

			/* drop from the list */
			write_lock(&endpoint->pull_handles_lock);
			list_del(&handle->list_elt);
			write_unlock(&endpoint->pull_handles_lock);

			spin_unlock_bh(&handle->lock);
			/* release the timer reference */
			omx_pull_handle_release(handle);

		} else {
			dprintk(PULL, "del_timer_sync was useless pull handle %p timer, already exited\n", handle);

			/* the timer expired, nothing to do */
			BUG_ON(handle->status != OMX_PULL_HANDLE_STATUS_TIMER_EXITED);

			spin_unlock_bh(&handle->lock);
		}

		omx_pull_handle_release(handle);
		write_lock_bh(&endpoint->pull_handles_lock);
	}
	write_unlock_bh(&endpoint->pull_handles_lock);
}

/******************************
 * Endpoint pull-magic management
 */

/* the magic is an incremented 24bits generation followed by 8bits for the endpoint index */
#define OMX_PULL_HANDLE_MAGIC_ENDPOINT_INDEX_BITS 8
#if OMX_ENDPOINT_INDEX_MAX > (1<<OMX_PULL_HANDLE_MAGIC_ENDPOINT_INDEX_BITS)
#error Endpoint index may not fit in the pull handle magic
#endif

#define OMX_PULL_HANDLE_MAGIC_XOR 0x21071980

static uint32_t omx_endpoint_magic_generation = 0;

static INLINE uint32_t
omx_generate_pull_magic(struct omx_endpoint * endpoint)
{
	omx_endpoint_magic_generation++;
	return ((omx_endpoint_magic_generation << OMX_PULL_HANDLE_MAGIC_ENDPOINT_INDEX_BITS)
		| endpoint->endpoint_index)
	       ^ OMX_PULL_HANDLE_MAGIC_XOR;
}

/*
 * Acquire an endpoint using a pull handle magic given on the wire.
 *
 * Returns an endpoint acquired, on ERR_PTR(-errno) on error
 */
static INLINE struct omx_endpoint *
omx_endpoint_acquire_by_pull_magic(struct omx_iface * iface, uint32_t magic)
{
	uint8_t index = (magic ^ OMX_PULL_HANDLE_MAGIC_XOR) & ((1UL << OMX_PULL_HANDLE_MAGIC_ENDPOINT_INDEX_BITS) - 1);
	return omx_endpoint_acquire_by_iface_index(iface, index);
}

/************************
 * Pull handles creation
 */

static INLINE int
omx_pull_handle_pkt_hdr_fill(struct omx_endpoint * endpoint,
			     struct omx_pull_handle * handle,
			     struct omx_cmd_pull * cmd)
{
	struct omx_iface * iface = endpoint->iface;
	struct net_device * ifp = iface->eth_ifp;
	struct omx_hdr * mh = &handle->pkt_hdr;
	struct omx_pkt_head * ph = &mh->head;
	struct ethhdr * eh = &ph->eth;
	struct omx_pkt_pull_request * pull_n = &mh->body.pull;
	int ret;

	/* pre-fill the packet header */
	eh->h_proto = __constant_cpu_to_be16(ETH_P_OMX);
	memcpy(eh->h_source, ifp->dev_addr, sizeof (eh->h_source));

	/* set destination peer */
	ret = omx_set_target_peer(ph, cmd->peer_index);
	if (ret < 0) {
		printk(KERN_INFO "Open-MX: Failed to fill target peer in pull request header\n");
		goto out;
	}

	/* fill omx header */
	OMX_PKT_FIELD_FROM(pull_n->ptype, OMX_PKT_TYPE_PULL);
	OMX_PKT_FIELD_FROM(pull_n->src_endpoint, endpoint->endpoint_index);
	OMX_PKT_FIELD_FROM(pull_n->dst_endpoint, cmd->dest_endpoint);
	OMX_PKT_FIELD_FROM(pull_n->session, cmd->session_id);
	OMX_PKT_FIELD_FROM(pull_n->total_length, handle->total_length);
	OMX_PKT_FIELD_FROM(pull_n->pulled_rdma_id, cmd->remote_rdma_id);
	OMX_PKT_FIELD_FROM(pull_n->pulled_rdma_seqnum, cmd->remote_rdma_seqnum);
	OMX_PKT_FIELD_FROM(pull_n->pulled_rdma_offset, handle->pulled_rdma_offset);
	OMX_PKT_FIELD_FROM(pull_n->src_pull_handle, handle->idr_index);
	OMX_PKT_FIELD_FROM(pull_n->src_magic, handle->magic);

	/* block_length, frame_index, and first_frame_offset filled at actual send */

	return 0;

 out:
	return ret;
}

/*
 * Create a pull handle and return it as acquired and locked.
 */
static INLINE struct omx_pull_handle *
omx_pull_handle_create(struct omx_endpoint * endpoint,
		       struct omx_cmd_pull * cmd)
{
	struct omx_pull_handle * handle;
	struct omx_user_region * region;
	int i;
	int err;

	/* acquire the region */
	region = omx_user_region_acquire(endpoint, cmd->local_rdma_id);
	if (unlikely(!region)) {
		err = -ENOMEM;
		goto out;
	}

	/* alloc the pull handle */
	handle = kmalloc(sizeof(struct omx_pull_handle), GFP_KERNEL);
	if (unlikely(!handle)) {
		printk(KERN_INFO "Open-MX: Failed to allocate a pull handle\n");
		goto out_with_region;
	}
	/* initialize the lock, we will acquire it soon */
	spin_lock_init(&handle->lock);

	/* while failed, realloc and retry */
 idr_try_alloc:
	err = idr_pre_get(&endpoint->pull_handles_idr, GFP_KERNEL);
	if (unlikely(!err)) {
		printk(KERN_ERR "Open-MX: Failed to allocate idr space for pull handles\n");
		err = -ENOMEM;
		goto out_with_handle;
	}

	/* lock the handle lock first we'll need it later and can't take it after the endpoint lock */
	spin_lock(&handle->lock);

	write_lock_bh(&endpoint->pull_handles_lock);
	err = idr_get_new(&endpoint->pull_handles_idr, handle, &handle->idr_index);
	if (unlikely(err == -EAGAIN)) {
		write_unlock_bh(&endpoint->pull_handles_lock);
		spin_unlock(&handle->lock);
		goto idr_try_alloc;
	}

	/* we are good now, finish filling the handle */
	kref_init(&handle->refcount);
	handle->endpoint = endpoint;
	handle->region = region;
	handle->total_length = cmd->length;
	handle->puller_rdma_offset = cmd->local_offset;
	handle->pulled_rdma_offset = cmd->remote_offset;

	/* initialize variable stuff */
	handle->status = OMX_PULL_HANDLE_STATUS_OK;
	handle->remaining_length = cmd->length;
	handle->frame_index = 0;
	handle->next_frame_index = 0;
	handle->nr_requested_frames = 0;
	handle->nr_missing_frames = 0;
	handle->nr_copying_frames = 0;
	handle->nr_valid_block_descs = 0;
	for(i=0; i<OMX_PULL_BLOCK_DESCS_NR-1; i++)
		handle->block_desc[i].frames_missing_bitmap = 0; /* make sure the invalid block descs are easy to check */
	handle->already_rerequested_blocks = 0;
	handle->last_retransmit_jiffies = cmd->resend_timeout_jiffies + jiffies;
	handle->magic = omx_generate_pull_magic(endpoint);

#ifdef CONFIG_NET_DMA
	handle->dma_chan = NULL;
	handle->dma_last_cookie = -1;
	skb_queue_head_init(&handle->dma_skb_queue);
#endif

	/* initialize the completion event */
	handle->done_event.local_rdma_id = cmd->local_rdma_id;
	handle->done_event.lib_cookie = cmd->lib_cookie;

	/* initialize cached header */
	err = omx_pull_handle_pkt_hdr_fill(endpoint, handle, cmd);
	if (err < 0)
		goto out_with_idr;

	/* init timer */
	setup_timer(&handle->retransmit_timer, omx_pull_handle_timeout_handler,
		    (unsigned long) handle);

	/* queue in the endpoint list */
	list_add_tail(&handle->list_elt,
		      &endpoint->pull_handles_running_list);

	write_unlock_bh(&endpoint->pull_handles_lock);

	dprintk(PULL, "created and acquired pull handle %p\n", handle);

	/* tell the sparse checker that we keep the lock and pass it back to the caller */
	__release(&handle->lock);

	return handle;

 out_with_idr:
	idr_remove(&endpoint->pull_handles_idr, handle->idr_index);
	write_unlock_bh(&endpoint->pull_handles_lock);
	spin_unlock(&handle->lock);
 out_with_handle:
	kfree(handle);
 out_with_region:
	omx_user_region_release(region);
 out:
	return NULL;
}

/*************************
 * Pull handle completion
 */

static INLINE void
omx_pull_handle_complete_notify(struct omx_pull_handle * handle)
{
	struct omx_endpoint * endpoint = handle->endpoint;

	omx_notify_exp_event(endpoint,
			     OMX_EVT_PULL_DONE,
			     &handle->done_event, sizeof(handle->done_event));

	/* release the handle */
	omx_pull_handle_release(handle);

	/*
	 * do not release the region here, let the last pull user release it.
	 * if weare completing the pull with an error, there could be other users in memcpy
	 */
}

/*
 * Takes an acquired and locked pull handle and complete it.
 * Called by the BH after receiving a pull reply or a nack,
 * or by the retransmit timer when expired.
 *
 * If the timeout expired, status is OMX_EVT_PULL_DONE_TIMEOUT
 * and the timer will exit right after returning from here.
 * In the other cases, the timer needs to catch TIMER_MUST_EXIT.
 */
static INLINE void
omx_pull_handle_complete(struct omx_pull_handle * handle, uint8_t status)
{
	struct omx_endpoint * endpoint = handle->endpoint;
	int ret;

	/* tell the sparse checker that the caller took the lock */
	__acquire(&handle->lock);

	BUG_ON(handle->status != OMX_PULL_HANDLE_STATUS_OK);
	handle->status = (status == OMX_EVT_PULL_DONE_TIMEOUT)
			? OMX_PULL_HANDLE_STATUS_TIMER_EXITED
			: OMX_PULL_HANDLE_STATUS_TIMER_MUST_EXIT;

	/* remove from the idr (and endpoint list) so that no incoming packet can find it anymore */
	write_lock_bh(&endpoint->pull_handles_lock);
	idr_remove(&endpoint->pull_handles_idr, handle->idr_index);
	if (status == OMX_EVT_PULL_DONE_TIMEOUT) {
		dprintk(PULL, "pull handle %p timer done, removing from idr and endpoint list\n", handle);
		list_del(&handle->list_elt);
	} else {
		dprintk(PULL, "moving done handle %p to the done_but_timer list and removing from idr\n", handle);
		list_move(&handle->list_elt, &endpoint->pull_handles_done_but_timer_list);
	}
	write_unlock_bh(&endpoint->pull_handles_lock);

	spin_unlock(&handle->lock);

	/* now that the handle cannot be used by anybody else, complete it for real */

	/* finish filling the event for user-space */
	handle->done_event.status = status;
	handle->done_event.pulled_length = handle->total_length - handle->remaining_length;

	/* see if the offloaded copies are done */
	ret = omx_pull_handle_deferred_wait_dma_completions(handle);
	if (!ret)
		omx_pull_handle_complete_notify(handle);
}

/************************
 * Sending pull requests
 */

/* Called with the handle acquired and locked */
static INLINE struct sk_buff *
omx_fill_pull_block_request(struct omx_pull_handle * handle, int desc_nr)
{
	struct omx_pull_block_desc * desc = &handle->block_desc[desc_nr];
	struct omx_iface * iface = handle->endpoint->iface;
	uint32_t frame_index = desc->frame_index;
	uint32_t block_length = desc->block_length;
	uint32_t first_frame_offset = desc->first_frame_offset;
	struct sk_buff * skb;
	struct omx_hdr * mh;
	struct omx_pkt_pull_request * pull_n;
	size_t hdr_len = sizeof(struct omx_pkt_head) + sizeof(struct omx_pkt_pull_request);

	skb = omx_new_skb(/* pad to ETH_ZLEN */
			  max_t(unsigned long, hdr_len, ETH_ZLEN));
	if (unlikely(skb == NULL)) {
		omx_counter_inc(iface, SEND_NOMEM_SKB);
		printk(KERN_INFO "Open-MX: Failed to create pull skb\n");
		return ERR_PTR(-ENOMEM);
	}

	/* locate headers */
	mh = omx_skb_mac_header(skb);
	pull_n = &mh->body.pull;

	/* copy common pkt hdrs from the handle */
	memcpy(mh, &handle->pkt_hdr, sizeof(handle->pkt_hdr));

	OMX_PKT_FIELD_FROM(pull_n->block_length, block_length);
	OMX_PKT_FIELD_FROM(pull_n->first_frame_offset, first_frame_offset);
	OMX_PKT_FIELD_FROM(pull_n->frame_index, frame_index);

	omx_send_dprintk(&mh->head.eth, "PULL handle %lx magic %lx length %ld out of %ld, frame index %ld first_frame_offset %ld",
			 (unsigned long) OMX_FROM_PKT_FIELD(pull_n->src_pull_handle),
			 (unsigned long) OMX_FROM_PKT_FIELD(pull_n->src_magic),
			 (unsigned long) block_length,
			 (unsigned long) OMX_FROM_PKT_FIELD(pull_n->total_length),
			 (unsigned long) frame_index,
			 (unsigned long) first_frame_offset);

	return skb;
}

int
omx_ioctl_pull(struct omx_endpoint * endpoint,
	       void __user * uparam)
{
	struct omx_cmd_pull cmd;
	struct omx_pull_handle * handle;
	struct omx_iface * iface = endpoint->iface;
	struct sk_buff * skb, * skbs[OMX_PULL_BLOCK_DESCS_NR] = { NULL };
	uint32_t block_length;
	int i;
	int err = 0;

	err = copy_from_user(&cmd, uparam, sizeof(cmd));
	if (unlikely(err != 0)) {
		printk(KERN_ERR "Open-MX: Failed to read send pull cmd hdr\n");
		err = -EFAULT;
		goto out;
	}

	/* check the offsets */
	if (cmd.local_offset >= OMX_PULL_REPLY_LENGTH_MAX
	    || cmd.remote_offset >= OMX_PULL_REPLY_LENGTH_MAX) {
		err = -EINVAL;
		goto out;
	}

#ifndef OMX_DISABLE_SHARED
	if (unlikely(cmd.shared))
		return omx_shared_pull(endpoint, &cmd);
#endif

	/* create, acquire and lock the handle */
	handle = omx_pull_handle_create(endpoint, &cmd);
	if (unlikely(!handle)) {
		printk(KERN_INFO "Open-MX: Failed to allocate a pull handle\n");
		err = -ENOMEM;
		goto out;
	}

	/* tell the sparse checker that the lock has been taken by omx_pull_handle_create() */
	__acquire(&handle->lock);

	/* send a first pull block request */
	block_length = OMX_PULL_BLOCK_LENGTH_MAX
		- handle->pulled_rdma_offset;
	if (block_length > handle->remaining_length)
		block_length = handle->remaining_length;

	omx_pull_handle_append_needed_frames(handle, block_length, handle->pulled_rdma_offset);
	skb = omx_fill_pull_block_request(handle, 0);
	if (unlikely(IS_ERR(skb))) {
		BUG_ON(PTR_ERR(skb) != -ENOMEM);
		/* just ignore the memory allocation failure and let retransmission take care of it */
	} else {
		skbs[0] = skb;
	}

	for(i=1; i<OMX_PULL_BLOCK_DESCS_NR; i++) {
		/* send another pull block request if needed */

		if (!handle->remaining_length)
			break;

		block_length = OMX_PULL_BLOCK_LENGTH_MAX;
		if (block_length > handle->remaining_length)
			block_length = handle->remaining_length;

		omx_pull_handle_append_needed_frames(handle, block_length, 0);
		skb = omx_fill_pull_block_request(handle, i);
		if (unlikely(IS_ERR(skb))) {
			BUG_ON(PTR_ERR(skb) != -ENOMEM);
			/* just ignore the memory allocation failure and let retransmission take care of it */
		} else {
			skbs[i] = skb;
		}
	}

	/* schedule the timeout handler now that we are ready to send the requests */
	__mod_timer(&handle->retransmit_timer,
		    jiffies + OMX_PULL_RETRANSMIT_TIMEOUT_JIFFIES);

	/*
	 * do not keep the lock while sending
	 * since the loopback device may cause reentrancy
	 */
	spin_unlock(&handle->lock);

	for(i=0; i<OMX_PULL_BLOCK_DESCS_NR; i++)
		if (likely(skbs[i]))
			omx_queue_xmit(iface, skbs[i], PULL_REQ);

	return 0;

 out:
	return err;
}

/*************************
 * Handle timeout handler
 */

static INLINE void
omx_progress_pull_on_handle_timeout_handle_locked(struct omx_iface * iface,
						  struct omx_pull_handle * handle)
{
	struct sk_buff *skb, *skbs[OMX_PULL_BLOCK_DESCS_NR] = { NULL };
	int i;

	/* tell the sparse checker that the lock has been taken by the caller */
	__acquire(&handle->lock);

	/* request the first block again */
	omx_counter_inc(iface, PULL_TIMEOUT_HANDLER_FIRST_BLOCK);

	skb = omx_fill_pull_block_request(handle, 0);
	if (unlikely(IS_ERR(skb))) {
		BUG_ON(PTR_ERR(skb) != -ENOMEM);
	} else {
		handle->already_rerequested_blocks = 0;
		skbs[0] = skb;
	}

	/*
	 * If the other blocks aren't done either, request it again
	 * (otherwise the N-block pipeline would be broken for ever)
	 * This shouldn't happen often since it means a packet has been lost
	 * in each block.
	 */
	for(i=1; i<OMX_PULL_BLOCK_DESCS_NR; i++) {
		if (handle->block_desc[i].frames_missing_bitmap) {
			omx_counter_inc(iface, PULL_TIMEOUT_HANDLER_NONFIRST_BLOCK);

			skb = omx_fill_pull_block_request(handle, i);
			if (unlikely(IS_ERR(skb))) {
				BUG_ON(PTR_ERR(skb) != -ENOMEM);
			} else {
				skbs[i] = skb;
			}
		}
	}

	/* cleanup a bit of dma-offloaded copies */
	omx_pull_handle_poll_dma_completions(handle);

	/* reschedule another timeout handler */
	mod_timer(&handle->retransmit_timer,
		  jiffies + OMX_PULL_RETRANSMIT_TIMEOUT_JIFFIES);

	/*
	 * do not keep the lock while sending
	 * since the loopback device may cause reentrancy<commit_msg>Whenever we fail to send a pull request, don't try to send the ones for the other blocks since they will very likely fail too. Just let retransmission occur later.<commit_after>/*
 * Open-MX
 * Copyright © INRIA 2007-2008 (see AUTHORS file)
 *
 * The development of this software has been funded by Myricom, Inc.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or (at
 * your option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 *
 * See the GNU General Public License in COPYING.GPL for more details.
 */

#include <linux/kernel.h>
#include <linux/slab.h>
#include <linux/spinlock.h>
#include <linux/idr.h>
#include <linux/kref.h>
#include <linux/timer.h>
#include <linux/workqueue.h>

#include "omx_misc.h"
#include "omx_hal.h"
#include "omx_wire_access.h"
#include "omx_common.h"
#include "omx_iface.h"
#include "omx_peer.h"
#include "omx_endpoint.h"
#include "omx_reg.h"
#include "omx_dma.h"
#ifndef OMX_DISABLE_SHARED
#include "omx_shared.h"
#endif

/**************************
 * Pull-specific Constants
 */

#define OMX_PULL_BLOCK_LENGTH_MAX (OMX_PULL_REPLY_LENGTH_MAX*OMX_PULL_REPLY_PER_BLOCK)

#define OMX_PULL_RETRANSMIT_TIMEOUT_MS	1000
#define OMX_PULL_RETRANSMIT_TIMEOUT_JIFFIES (OMX_PULL_RETRANSMIT_TIMEOUT_MS*HZ/1000)

#define OMX_PULL_BLOCK_DESCS_NR 2

#ifdef OMX_MX_WIRE_COMPAT
#if OMX_PULL_REPLY_LENGTH_MAX >= 65536
#error Cannot store rdma offsets > 65535 in 16bits offsets on the wire
#endif
#endif

/**********************
 * Pull-specific Types
 */

#if OMX_PULL_REPLY_PER_BLOCK & (OMX_PULL_REPLY_PER_BLOCK-1)
/* we don't want to divide by a non-power-of-two */
#error Need a power of two as the number of replies per pull block
#elif OMX_PULL_REPLY_PER_BLOCK > 64
#error Cannot request more than 64 replies per pull block
#elif OMX_PULL_REPLY_PER_BLOCK > 32
typedef uint64_t omx_block_frame_bitmask_t;
#elif OMX_PULL_REPLY_PER_BLOCK > 16
typedef uint32_t omx_block_frame_bitmask_t;
#elif OMX_PULL_REPLY_PER_BLOCK > 8
typedef uint16_t omx_block_frame_bitmask_t;
#else
typedef uint8_t omx_block_frame_bitmask_t;
#endif

enum omx_pull_handle_status {
	/*
	 * The handle is normal, being processed as usual and its timeout handler is running ok.
	 * It is queued on the endpoint running_list.
	 */
	OMX_PULL_HANDLE_STATUS_OK,

	/*
	 * The handle has been removed from the idr, but the timeout handler is still running.
	 * It is queued on the endpoint done_but_timer_list.
	 * Either the pull has completed (or aborted on error), or the endpoint is being closed and
	 * all handles have been scheduled for removal.
	 * The timeout handler must exit next time it runs. It will release the reference on the handle,
	 * dequeue it, and the handle may be destroyed.
	 */
	OMX_PULL_HANDLE_STATUS_TIMER_MUST_EXIT,

	/*
	 * The handle has been removed from the idr and the endpoint lists, and its timeout handler has exited
	 * and released its reference.
	 * Either the pull has completed (or aborted on error) and the handler caught the MUST_EXIT status,
	 * or the timeout was reached and its handler aborted the handle directly.
	 */
	OMX_PULL_HANDLE_STATUS_TIMER_EXITED,
};

struct omx_pull_block_desc {
	uint32_t frame_index;
	uint32_t block_length;
	uint32_t first_frame_offset;
	omx_block_frame_bitmask_t frames_missing_bitmap; /* frames not received at all */
};

struct omx_pull_handle {
	struct kref refcount;
	struct list_head list_elt; /* always queued on one of the endpoint lists */
	int idr_index; /* that's what idr_get_new wants */
	uint32_t magic; /* 8bits endpoint index + 24bits handle generation number */

	/* timer for retransmission */
	struct timer_list retransmit_timer;
	uint64_t last_retransmit_jiffies;

	/* global pull fields */
	struct omx_endpoint * endpoint;
	struct omx_user_region * region;
	uint32_t total_length;
	uint32_t puller_rdma_offset;
	uint32_t pulled_rdma_offset;

	/* current status */
	spinlock_t lock;
	enum omx_pull_handle_status status;
	uint32_t remaining_length;
	uint32_t frame_index; /* index of the first requested frame */
	uint32_t next_frame_index; /* index of the frame to request */
	uint32_t nr_requested_frames; /* number of frames requested */
	uint32_t nr_missing_frames; /* frames requested but not received yet */
	uint32_t nr_copying_frames; /* frames received but not copied yet*/
	uint32_t nr_valid_block_descs;
	uint32_t already_rerequested_blocks; /* amount of first blocks that were requested again since the last timer */
	struct omx_pull_block_desc block_desc[OMX_PULL_BLOCK_DESCS_NR];

#ifdef CONFIG_NET_DMA
	struct dma_chan *dma_chan;
	dma_cookie_t dma_last_cookie;
	struct sk_buff_head dma_skb_queue;
	struct work_struct deferred_dma_wait_work;
#endif

	/* completion event */
	struct omx_evt_pull_done done_event;

	/* pull packet header */
	struct omx_hdr pkt_hdr;
};

static void omx_pull_handle_timeout_handler(unsigned long data);

#ifdef CONFIG_NET_DMA
static void omx_pull_handle_poll_dma_completions(struct omx_pull_handle *handle);
static int omx_pull_handle_deferred_wait_dma_completions(struct omx_pull_handle *handle);
#else
#define omx_pull_handle_poll_dma_completions(ph) /* nothing */
#define omx_pull_handle_deferred_wait_dma_completions(ph) 0 /* always completed */
#endif

/*
 * Notes about locking:
 *
 * Each handle owns a spin_lock that protects the actual pull status (frame index, ...).
 * It also protects its handle status and its queueing in the endpoint lists and idr.
 * This lock is always taken *before* the endpoint pull handle lock.
 *
 * The handle does not own a reference on the endpoint. It is always queued in one
 * endpoint lists, and the endpoint closing will enforce its detruction.
 * When the endpoint starts to be closed, it calls prepare_exit which sets all handle
 * to timer_must_exit but cannot wait for them because of the possible interrupt
 * context. Later, the cleanup thread will cleanup the endpoint, including destroying
 * the handle timers that are still running.
 *
 * The pile of handles for an endpoint is protected by a rwlock. It is taken for
 * reading when acquiring an handle (when a pull reply or nack mcp arrives, likely
 * in a bottom half). It is taken for writing when creating a handle (when the
 * application request a pull), finishing a handle (when a pull reply arrives
 * and completes the pull request, likely in a bottom half), and when destroying
 * and remaining handles (when an endpoint is closed).
 * Since a bottom half and the application may both acquire the rwlock for
 * writing, we must always disable bottom halves when taking the rwlock for
 * either read or writing.
 */

/*
 * Notes about retransmission:
 *
 * The puller request 2 blocks of data, and waits for OMX_PULL_REPLY_PER_BLOCK
 * replies for each of them.
 *
 * A timer is set to detect when nothing has been received for a while.  It is
 * updated every time a new reply is received. This timer repost requests to
 * get current blocks (using descriptors that were cached in the pull handle).
 *
 * Additionally, if the second block completes before the first one, there is
 * a good chance that one packet got lost for the first block. In this case,
 * we optimistically re-request the first block.
 * To avoid re-requesting too often, we do it only once per timeout.
 *
 * In the end, the timer is only called if:
 * + one packet is lost for both the first and the second current block
 * + or one packet is missing in the first block after one optimistic re-request.
 * So the timeout doesn't need to be short, 1 second is enough.
 */

#ifdef OMX_DRIVER_DEBUG
/* defined as module parameters */
extern unsigned long omx_PULL_REQ_packet_loss;
extern unsigned long omx_PULL_REPLY_packet_loss;
/* index between 0 and the above limit */
static unsigned long omx_PULL_REQ_packet_loss_index = 0;
static unsigned long omx_PULL_REPLY_packet_loss_index = 0;
#endif /* OMX_DRIVER_DEBUG */

/********************************
 * Pull handle bitmap management
 */

static INLINE void
omx_pull_handle_append_needed_frames(struct omx_pull_handle * handle,
				     uint32_t block_length,
				     uint32_t first_frame_offset)
{
	struct omx_pull_block_desc *desc;
	omx_block_frame_bitmask_t new_mask;
	int new_frames;

	new_frames = (first_frame_offset + block_length
		      + OMX_PULL_REPLY_LENGTH_MAX-1) / OMX_PULL_REPLY_LENGTH_MAX;
	new_mask = ((omx_block_frame_bitmask_t) -1) >> (OMX_PULL_REPLY_PER_BLOCK-new_frames);

	desc = &handle->block_desc[handle->nr_valid_block_descs];
	desc->frame_index = handle->next_frame_index;
	desc->block_length = block_length;
	desc->first_frame_offset = first_frame_offset;
	desc->frames_missing_bitmap = new_mask;

	handle->nr_requested_frames += new_frames;
	handle->nr_missing_frames += new_frames;
	handle->next_frame_index += new_frames;
	handle->remaining_length -= block_length;
	handle->nr_valid_block_descs++;

	dprintk(PULL, "appending block #%d with %d new frames to pull handle %p, now requested %ld-%ld\n",
		handle->nr_valid_block_descs-1, new_frames, handle,
		(unsigned long) handle->frame_index, (unsigned long) handle->next_frame_index-1);
}

static INLINE void
omx_pull_handle_first_block_done(struct omx_pull_handle * handle)
{
	uint32_t first_block_frames = min(handle->nr_requested_frames, (uint32_t) OMX_PULL_REPLY_PER_BLOCK);

	handle->frame_index += first_block_frames;
	handle->nr_requested_frames -= first_block_frames;
	handle->nr_valid_block_descs--;
	if (handle->already_rerequested_blocks)
		handle->already_rerequested_blocks--;
	memmove(&handle->block_desc[0], &handle->block_desc[1],
		sizeof(struct omx_pull_block_desc) * handle->nr_valid_block_descs);
	handle->block_desc[OMX_PULL_BLOCK_DESCS_NR-1].frames_missing_bitmap = 0; /* make sure the invalid block descs are easy to check */

	dprintk(PULL, "first block of pull handle %p done, removing %d requested frames, now requested %ld-%ld\n",
		handle, first_block_frames,
		(unsigned long) handle->frame_index, (unsigned long) handle->next_frame_index-1);
}

/**********************************
 * Pull handle acquiring/releasing
 */

/*
 * Acquire a handle.
 *
 * Either another reference on this handle should be owned,
 * or the endpoint lock should be hold.
 */
static INLINE void
omx_pull_handle_acquire(struct omx_pull_handle * handle)
{
	kref_get(&handle->refcount);
}

/*
 * Actual freeing of a handle when the last reference is released
 */
static void
__omx_pull_handle_last_release(struct kref * kref)
{
	struct omx_pull_handle * handle = container_of(kref, struct omx_pull_handle, refcount);

	dprintk(KREF, "releasing the last reference on pull handle %p\n",
		handle);

	BUG_ON(handle->status != OMX_PULL_HANDLE_STATUS_TIMER_EXITED);

	/* release the region now that we are sure that nobody else uses it */
	omx_user_region_release(handle->region);

	kfree(handle);
}

/*
 * Release an acquired pull handle
 */
static INLINE void
omx_pull_handle_release(struct omx_pull_handle * handle)
{
	kref_put(&handle->refcount, __omx_pull_handle_last_release);
}


/***************************************
 * Per-endpoint pull handles management
 */

int
omx_endpoint_pull_handles_init(struct omx_endpoint * endpoint)
{
	INIT_LIST_HEAD(&endpoint->pull_handles_running_list);
	INIT_LIST_HEAD(&endpoint->pull_handles_done_but_timer_list);
	idr_init(&endpoint->pull_handles_idr);
	rwlock_init(&endpoint->pull_handles_lock);
	return 0;
}

/*
 * Called when the last reference on the endpoint is removed,
 * possibly from unsafe context, cannot del_timer_sync() then.
 */
void
omx_endpoint_pull_handles_prepare_exit(struct omx_endpoint * endpoint)
{
	/*
	 * ask all pull handles of the endpoint to stop their timer.
	 * but we can't take endpoint->pull_handles_lock before handle->lock since that would deadlock
	 * so we use a tricky loop to take locks in order
	 */

	write_lock_bh(&endpoint->pull_handles_lock);
	while (!list_empty(&endpoint->pull_handles_running_list)) {
		struct omx_pull_handle * handle;

		/* get the first handle of the list, acquire it and release the list lock */
		handle = list_first_entry(&endpoint->pull_handles_running_list, struct omx_pull_handle, list_elt);
		omx_pull_handle_acquire(handle);
		write_unlock_bh(&endpoint->pull_handles_lock);

		/* take the handle lock and check the status in case it changed while the lock was released */
		spin_lock_bh(&handle->lock);
		if (handle->status == OMX_PULL_HANDLE_STATUS_OK) {
			/* the handle didn't change, do our stuff */
			handle->status = OMX_PULL_HANDLE_STATUS_TIMER_MUST_EXIT;

			/*
			 * remove from the idr so that no incoming packet can find it anymore,
			 * and move to the done_but_timer list
			 */
			dprintk(PULL, "moving handle %p to the done_but_timer list and removing from idr\n", handle);
			write_lock(&endpoint->pull_handles_lock);
			idr_remove(&endpoint->pull_handles_idr, handle->idr_index);
			list_move(&handle->list_elt, &endpoint->pull_handles_done_but_timer_list);
			write_unlock(&endpoint->pull_handles_lock);
		} else {
			/* the handle has been moved out of the list, it means that the timer is done, nothing to done */
		}
		spin_unlock_bh(&handle->lock);
		omx_pull_handle_release(handle);

		write_lock_bh(&endpoint->pull_handles_lock);
	}
	write_unlock_bh(&endpoint->pull_handles_lock);

	idr_destroy(&endpoint->pull_handles_idr);
}

/*
 * Called when cleaning the endpoint, always from the cleanup thread, may del_timer_sync() then.
 */
void
omx_endpoint_pull_handles_force_exit(struct omx_endpoint * endpoint)
{
	might_sleep();

	write_lock_bh(&endpoint->pull_handles_lock);
	while (!list_empty(&endpoint->pull_handles_done_but_timer_list)) {
		struct omx_pull_handle * handle;
		int ret;

		/* get the first handle of the list, acquire it and release the list lock */
		handle = list_first_entry(&endpoint->pull_handles_done_but_timer_list, struct omx_pull_handle, list_elt);
		omx_pull_handle_acquire(handle);
		write_unlock_bh(&endpoint->pull_handles_lock);

		dprintk(PULL, "stopping handle %p timer with del_sync_timer\n", handle);
		ret = del_timer_sync(&handle->retransmit_timer);
		spin_lock_bh(&handle->lock);
		if (ret) {
			dprintk(PULL, "del_timer_sync stopped pull handle %p timer\n", handle);

			/* we deactivated the timer, cleanup ourself */
			BUG_ON(handle->status != OMX_PULL_HANDLE_STATUS_TIMER_MUST_EXIT);
			handle->status = OMX_PULL_HANDLE_STATUS_TIMER_EXITED;

			/* drop from the list */
			write_lock(&endpoint->pull_handles_lock);
			list_del(&handle->list_elt);
			write_unlock(&endpoint->pull_handles_lock);

			spin_unlock_bh(&handle->lock);
			/* release the timer reference */
			omx_pull_handle_release(handle);

		} else {
			dprintk(PULL, "del_timer_sync was useless pull handle %p timer, already exited\n", handle);

			/* the timer expired, nothing to do */
			BUG_ON(handle->status != OMX_PULL_HANDLE_STATUS_TIMER_EXITED);

			spin_unlock_bh(&handle->lock);
		}

		omx_pull_handle_release(handle);
		write_lock_bh(&endpoint->pull_handles_lock);
	}
	write_unlock_bh(&endpoint->pull_handles_lock);
}

/******************************
 * Endpoint pull-magic management
 */

/* the magic is an incremented 24bits generation followed by 8bits for the endpoint index */
#define OMX_PULL_HANDLE_MAGIC_ENDPOINT_INDEX_BITS 8
#if OMX_ENDPOINT_INDEX_MAX > (1<<OMX_PULL_HANDLE_MAGIC_ENDPOINT_INDEX_BITS)
#error Endpoint index may not fit in the pull handle magic
#endif

#define OMX_PULL_HANDLE_MAGIC_XOR 0x21071980

static uint32_t omx_endpoint_magic_generation = 0;

static INLINE uint32_t
omx_generate_pull_magic(struct omx_endpoint * endpoint)
{
	omx_endpoint_magic_generation++;
	return ((omx_endpoint_magic_generation << OMX_PULL_HANDLE_MAGIC_ENDPOINT_INDEX_BITS)
		| endpoint->endpoint_index)
	       ^ OMX_PULL_HANDLE_MAGIC_XOR;
}

/*
 * Acquire an endpoint using a pull handle magic given on the wire.
 *
 * Returns an endpoint acquired, on ERR_PTR(-errno) on error
 */
static INLINE struct omx_endpoint *
omx_endpoint_acquire_by_pull_magic(struct omx_iface * iface, uint32_t magic)
{
	uint8_t index = (magic ^ OMX_PULL_HANDLE_MAGIC_XOR) & ((1UL << OMX_PULL_HANDLE_MAGIC_ENDPOINT_INDEX_BITS) - 1);
	return omx_endpoint_acquire_by_iface_index(iface, index);
}

/************************
 * Pull handles creation
 */

static INLINE int
omx_pull_handle_pkt_hdr_fill(struct omx_endpoint * endpoint,
			     struct omx_pull_handle * handle,
			     struct omx_cmd_pull * cmd)
{
	struct omx_iface * iface = endpoint->iface;
	struct net_device * ifp = iface->eth_ifp;
	struct omx_hdr * mh = &handle->pkt_hdr;
	struct omx_pkt_head * ph = &mh->head;
	struct ethhdr * eh = &ph->eth;
	struct omx_pkt_pull_request * pull_n = &mh->body.pull;
	int ret;

	/* pre-fill the packet header */
	eh->h_proto = __constant_cpu_to_be16(ETH_P_OMX);
	memcpy(eh->h_source, ifp->dev_addr, sizeof (eh->h_source));

	/* set destination peer */
	ret = omx_set_target_peer(ph, cmd->peer_index);
	if (ret < 0) {
		printk(KERN_INFO "Open-MX: Failed to fill target peer in pull request header\n");
		goto out;
	}

	/* fill omx header */
	OMX_PKT_FIELD_FROM(pull_n->ptype, OMX_PKT_TYPE_PULL);
	OMX_PKT_FIELD_FROM(pull_n->src_endpoint, endpoint->endpoint_index);
	OMX_PKT_FIELD_FROM(pull_n->dst_endpoint, cmd->dest_endpoint);
	OMX_PKT_FIELD_FROM(pull_n->session, cmd->session_id);
	OMX_PKT_FIELD_FROM(pull_n->total_length, handle->total_length);
	OMX_PKT_FIELD_FROM(pull_n->pulled_rdma_id, cmd->remote_rdma_id);
	OMX_PKT_FIELD_FROM(pull_n->pulled_rdma_seqnum, cmd->remote_rdma_seqnum);
	OMX_PKT_FIELD_FROM(pull_n->pulled_rdma_offset, handle->pulled_rdma_offset);
	OMX_PKT_FIELD_FROM(pull_n->src_pull_handle, handle->idr_index);
	OMX_PKT_FIELD_FROM(pull_n->src_magic, handle->magic);

	/* block_length, frame_index, and first_frame_offset filled at actual send */

	return 0;

 out:
	return ret;
}

/*
 * Create a pull handle and return it as acquired and locked.
 */
static INLINE struct omx_pull_handle *
omx_pull_handle_create(struct omx_endpoint * endpoint,
		       struct omx_cmd_pull * cmd)
{
	struct omx_pull_handle * handle;
	struct omx_user_region * region;
	int i;
	int err;

	/* acquire the region */
	region = omx_user_region_acquire(endpoint, cmd->local_rdma_id);
	if (unlikely(!region)) {
		err = -ENOMEM;
		goto out;
	}

	/* alloc the pull handle */
	handle = kmalloc(sizeof(struct omx_pull_handle), GFP_KERNEL);
	if (unlikely(!handle)) {
		printk(KERN_INFO "Open-MX: Failed to allocate a pull handle\n");
		goto out_with_region;
	}
	/* initialize the lock, we will acquire it soon */
	spin_lock_init(&handle->lock);

	/* while failed, realloc and retry */
 idr_try_alloc:
	err = idr_pre_get(&endpoint->pull_handles_idr, GFP_KERNEL);
	if (unlikely(!err)) {
		printk(KERN_ERR "Open-MX: Failed to allocate idr space for pull handles\n");
		err = -ENOMEM;
		goto out_with_handle;
	}

	/* lock the handle lock first we'll need it later and can't take it after the endpoint lock */
	spin_lock(&handle->lock);

	write_lock_bh(&endpoint->pull_handles_lock);
	err = idr_get_new(&endpoint->pull_handles_idr, handle, &handle->idr_index);
	if (unlikely(err == -EAGAIN)) {
		write_unlock_bh(&endpoint->pull_handles_lock);
		spin_unlock(&handle->lock);
		goto idr_try_alloc;
	}

	/* we are good now, finish filling the handle */
	kref_init(&handle->refcount);
	handle->endpoint = endpoint;
	handle->region = region;
	handle->total_length = cmd->length;
	handle->puller_rdma_offset = cmd->local_offset;
	handle->pulled_rdma_offset = cmd->remote_offset;

	/* initialize variable stuff */
	handle->status = OMX_PULL_HANDLE_STATUS_OK;
	handle->remaining_length = cmd->length;
	handle->frame_index = 0;
	handle->next_frame_index = 0;
	handle->nr_requested_frames = 0;
	handle->nr_missing_frames = 0;
	handle->nr_copying_frames = 0;
	handle->nr_valid_block_descs = 0;
	for(i=0; i<OMX_PULL_BLOCK_DESCS_NR-1; i++)
		handle->block_desc[i].frames_missing_bitmap = 0; /* make sure the invalid block descs are easy to check */
	handle->already_rerequested_blocks = 0;
	handle->last_retransmit_jiffies = cmd->resend_timeout_jiffies + jiffies;
	handle->magic = omx_generate_pull_magic(endpoint);

#ifdef CONFIG_NET_DMA
	handle->dma_chan = NULL;
	handle->dma_last_cookie = -1;
	skb_queue_head_init(&handle->dma_skb_queue);
#endif

	/* initialize the completion event */
	handle->done_event.local_rdma_id = cmd->local_rdma_id;
	handle->done_event.lib_cookie = cmd->lib_cookie;

	/* initialize cached header */
	err = omx_pull_handle_pkt_hdr_fill(endpoint, handle, cmd);
	if (err < 0)
		goto out_with_idr;

	/* init timer */
	setup_timer(&handle->retransmit_timer, omx_pull_handle_timeout_handler,
		    (unsigned long) handle);

	/* queue in the endpoint list */
	list_add_tail(&handle->list_elt,
		      &endpoint->pull_handles_running_list);

	write_unlock_bh(&endpoint->pull_handles_lock);

	dprintk(PULL, "created and acquired pull handle %p\n", handle);

	/* tell the sparse checker that we keep the lock and pass it back to the caller */
	__release(&handle->lock);

	return handle;

 out_with_idr:
	idr_remove(&endpoint->pull_handles_idr, handle->idr_index);
	write_unlock_bh(&endpoint->pull_handles_lock);
	spin_unlock(&handle->lock);
 out_with_handle:
	kfree(handle);
 out_with_region:
	omx_user_region_release(region);
 out:
	return NULL;
}

/*************************
 * Pull handle completion
 */

static INLINE void
omx_pull_handle_complete_notify(struct omx_pull_handle * handle)
{
	struct omx_endpoint * endpoint = handle->endpoint;

	omx_notify_exp_event(endpoint,
			     OMX_EVT_PULL_DONE,
			     &handle->done_event, sizeof(handle->done_event));

	/* release the handle */
	omx_pull_handle_release(handle);

	/*
	 * do not release the region here, let the last pull user release it.
	 * if weare completing the pull with an error, there could be other users in memcpy
	 */
}

/*
 * Takes an acquired and locked pull handle and complete it.
 * Called by the BH after receiving a pull reply or a nack,
 * or by the retransmit timer when expired.
 *
 * If the timeout expired, status is OMX_EVT_PULL_DONE_TIMEOUT
 * and the timer will exit right after returning from here.
 * In the other cases, the timer needs to catch TIMER_MUST_EXIT.
 */
static INLINE void
omx_pull_handle_complete(struct omx_pull_handle * handle, uint8_t status)
{
	struct omx_endpoint * endpoint = handle->endpoint;
	int ret;

	/* tell the sparse checker that the caller took the lock */
	__acquire(&handle->lock);

	BUG_ON(handle->status != OMX_PULL_HANDLE_STATUS_OK);
	handle->status = (status == OMX_EVT_PULL_DONE_TIMEOUT)
			? OMX_PULL_HANDLE_STATUS_TIMER_EXITED
			: OMX_PULL_HANDLE_STATUS_TIMER_MUST_EXIT;

	/* remove from the idr (and endpoint list) so that no incoming packet can find it anymore */
	write_lock_bh(&endpoint->pull_handles_lock);
	idr_remove(&endpoint->pull_handles_idr, handle->idr_index);
	if (status == OMX_EVT_PULL_DONE_TIMEOUT) {
		dprintk(PULL, "pull handle %p timer done, removing from idr and endpoint list\n", handle);
		list_del(&handle->list_elt);
	} else {
		dprintk(PULL, "moving done handle %p to the done_but_timer list and removing from idr\n", handle);
		list_move(&handle->list_elt, &endpoint->pull_handles_done_but_timer_list);
	}
	write_unlock_bh(&endpoint->pull_handles_lock);

	spin_unlock(&handle->lock);

	/* now that the handle cannot be used by anybody else, complete it for real */

	/* finish filling the event for user-space */
	handle->done_event.status = status;
	handle->done_event.pulled_length = handle->total_length - handle->remaining_length;

	/* see if the offloaded copies are done */
	ret = omx_pull_handle_deferred_wait_dma_completions(handle);
	if (!ret)
		omx_pull_handle_complete_notify(handle);
}

/************************
 * Sending pull requests
 */

/* Called with the handle acquired and locked */
static INLINE struct sk_buff *
omx_fill_pull_block_request(struct omx_pull_handle * handle, int desc_nr)
{
	struct omx_pull_block_desc * desc = &handle->block_desc[desc_nr];
	struct omx_iface * iface = handle->endpoint->iface;
	uint32_t frame_index = desc->frame_index;
	uint32_t block_length = desc->block_length;
	uint32_t first_frame_offset = desc->first_frame_offset;
	struct sk_buff * skb;
	struct omx_hdr * mh;
	struct omx_pkt_pull_request * pull_n;
	size_t hdr_len = sizeof(struct omx_pkt_head) + sizeof(struct omx_pkt_pull_request);

	skb = omx_new_skb(/* pad to ETH_ZLEN */
			  max_t(unsigned long, hdr_len, ETH_ZLEN));
	if (unlikely(skb == NULL)) {
		omx_counter_inc(iface, SEND_NOMEM_SKB);
		printk(KERN_INFO "Open-MX: Failed to create pull skb\n");
		return ERR_PTR(-ENOMEM);
	}

	/* locate headers */
	mh = omx_skb_mac_header(skb);
	pull_n = &mh->body.pull;

	/* copy common pkt hdrs from the handle */
	memcpy(mh, &handle->pkt_hdr, sizeof(handle->pkt_hdr));

	OMX_PKT_FIELD_FROM(pull_n->block_length, block_length);
	OMX_PKT_FIELD_FROM(pull_n->first_frame_offset, first_frame_offset);
	OMX_PKT_FIELD_FROM(pull_n->frame_index, frame_index);

	omx_send_dprintk(&mh->head.eth, "PULL handle %lx magic %lx length %ld out of %ld, frame index %ld first_frame_offset %ld",
			 (unsigned long) OMX_FROM_PKT_FIELD(pull_n->src_pull_handle),
			 (unsigned long) OMX_FROM_PKT_FIELD(pull_n->src_magic),
			 (unsigned long) block_length,
			 (unsigned long) OMX_FROM_PKT_FIELD(pull_n->total_length),
			 (unsigned long) frame_index,
			 (unsigned long) first_frame_offset);

	return skb;
}

int
omx_ioctl_pull(struct omx_endpoint * endpoint,
	       void __user * uparam)
{
	struct omx_cmd_pull cmd;
	struct omx_pull_handle * handle;
	struct omx_iface * iface = endpoint->iface;
	struct sk_buff * skb, * skbs[OMX_PULL_BLOCK_DESCS_NR] = { NULL };
	uint32_t block_length;
	int i;
	int err = 0;

	err = copy_from_user(&cmd, uparam, sizeof(cmd));
	if (unlikely(err != 0)) {
		printk(KERN_ERR "Open-MX: Failed to read send pull cmd hdr\n");
		err = -EFAULT;
		goto out;
	}

	/* check the offsets */
	if (cmd.local_offset >= OMX_PULL_REPLY_LENGTH_MAX
	    || cmd.remote_offset >= OMX_PULL_REPLY_LENGTH_MAX) {
		err = -EINVAL;
		goto out;
	}

#ifndef OMX_DISABLE_SHARED
	if (unlikely(cmd.shared))
		return omx_shared_pull(endpoint, &cmd);
#endif

	/* create, acquire and lock the handle */
	handle = omx_pull_handle_create(endpoint, &cmd);
	if (unlikely(!handle)) {
		printk(KERN_INFO "Open-MX: Failed to allocate a pull handle\n");
		err = -ENOMEM;
		goto out;
	}

	/* tell the sparse checker that the lock has been taken by omx_pull_handle_create() */
	__acquire(&handle->lock);

	/* send a first pull block request */
	block_length = OMX_PULL_BLOCK_LENGTH_MAX
		- handle->pulled_rdma_offset;
	if (block_length > handle->remaining_length)
		block_length = handle->remaining_length;

	omx_pull_handle_append_needed_frames(handle, block_length, handle->pulled_rdma_offset);
	skb = omx_fill_pull_block_request(handle, 0);
	if (unlikely(IS_ERR(skb))) {
		BUG_ON(PTR_ERR(skb) != -ENOMEM);
		/* just ignore the memory allocation failure and let retransmission take care of it */
		goto skbs_ready; /* don't try to submit more */
	} else {
		skbs[0] = skb;
	}

	for(i=1; i<OMX_PULL_BLOCK_DESCS_NR; i++) {
		/* send another pull block request if needed */

		if (!handle->remaining_length)
			break;

		block_length = OMX_PULL_BLOCK_LENGTH_MAX;
		if (block_length > handle->remaining_length)
			block_length = handle->remaining_length;

		omx_pull_handle_append_needed_frames(handle, block_length, 0);
		skb = omx_fill_pull_block_request(handle, i);
		if (unlikely(IS_ERR(skb))) {
			BUG_ON(PTR_ERR(skb) != -ENOMEM);
			/* just ignore the memory allocation failure and let retransmission take care of it */
			goto skbs_ready; /* don't try to submit more */
		} else {
			skbs[i] = skb;
		}
	}

 skbs_ready:
	/* schedule the timeout handler now that we are ready to send the requests */
	__mod_timer(&handle->retransmit_timer,
		    jiffies + OMX_PULL_RETRANSMIT_TIMEOUT_JIFFIES);

	/*
	 * do not keep the lock while sending
	 * since the loopback device may cause reentrancy
	 */
	spin_unlock(&handle->lock);

	for(i=0; i<OMX_PULL_BLOCK_DESCS_NR; i++)
		if (likely(skbs[i]))
			omx_queue_xmit(iface, skbs[i], PULL_REQ);

	return 0;

 out:
	return err;
}

/*************************
 * Handle timeout handler
 */

static INLINE void
omx_progress_pull_on_handle_timeout_handle_locked(struct omx_iface * iface,
						  struct omx_pull_handle * handle)
{
	struct sk_buff *skb, *skbs[OMX_PULL_BLOCK_DESCS_NR] = { NULL };
	int i;

	/* tell the sparse checker that the lock has been taken by the caller */
	__acquire(&handle->lock);

	/* request the first block again */
	omx_counter_inc(iface, PULL_TIMEOUT_HANDLER_FIRST_BLOCK);

	skb = omx_fill_pull_block_request(handle, 0);
	if (unlikely(IS_ERR(skb))) {
		BUG_ON(PTR_ERR(skb) != -ENOMEM);
		goto skbs_ready; /* don't try to submit more */
	} else {
		skbs[0] = skb;
		handle->already_rerequested_blocks = 0;
	}

	/*
	 * If the other blocks aren't done either, request it again
	 * (otherwise the N-block pipeline would be broken for ever)
	 * This shouldn't happen often since it means a packet has been lost
	 * in each block.
	 */
	for(i=1; i<OMX_PULL_BLOCK_DESCS_NR; i++) {
		if (handle->block_desc[i].frames_missing_bitmap) {
			omx_counter_inc(iface, PULL_TIMEOUT_HANDLER_NONFIRST_BLOCK);

			skb = omx_fill_pull_block_request(handle, i);
			if (unlikely(IS_ERR(skb))) {
				BUG_ON(PTR_ERR(skb) != -ENOMEM);
				goto skbs_ready; /* don't try to submit more */
			} else {
				skbs[i] = skb;
			}
		}
	}

 skbs_ready:
	/* cleanup a bit of dma-offloaded copies */
	omx_pull_handle_poll_dma_completions(handle);

	/* reschedule another timeout handler */
	mod_timer(&handle->retransmit_timer,
		  jiffies + OMX_PULL_RETRANSMIT_TIMEOUT_JIFFIES);

	/*
	 * do not keep the lock while sending
	 * since the loopback device may cause reentrancy
{'fraction_non_alphanumeric': 0.10213652944241793, 'fraction_numerical': 0.005614483349863, 'mean_word_length': 4.968696699107054, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 441, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4191138', 'n_tokens_mistral': 21230, 'n_tokens_neox': 20228, 'n_words': 5462}