ar-call.c

linux 内核源代码

}

/*
 * detach a call from a socket and set up for release
 */
void rxrpc_release_call(struct rxrpc_call *call)
{
	struct rxrpc_connection *conn = call->conn;
	struct rxrpc_sock *rx = call->socket;

	_enter("{%d,%d,%d,%d}",
	       call->debug_id, atomic_read(&call->usage),
	       atomic_read(&call->ackr_not_idle),
	       call->rx_first_oos);

	spin_lock_bh(&call->lock);
	if (test_and_set_bit(RXRPC_CALL_RELEASED, &call->flags))
		BUG();
	spin_unlock_bh(&call->lock);

	/* dissociate from the socket
	 * - the socket's ref on the call is passed to the death timer
	 */
	_debug("RELEASE CALL %p (%d CONN %p)", call, call->debug_id, conn);

	write_lock_bh(&rx->call_lock);
	if (!list_empty(&call->accept_link)) {
		_debug("unlinking once-pending call %p { e=%lx f=%lx }",
		       call, call->events, call->flags);
		ASSERT(!test_bit(RXRPC_CALL_HAS_USERID, &call->flags));
		list_del_init(&call->accept_link);
		sk_acceptq_removed(&rx->sk);
	} else if (test_bit(RXRPC_CALL_HAS_USERID, &call->flags)) {
		rb_erase(&call->sock_node, &rx->calls);
		memset(&call->sock_node, 0xdd, sizeof(call->sock_node));
		clear_bit(RXRPC_CALL_HAS_USERID, &call->flags);
	}
	write_unlock_bh(&rx->call_lock);

	/* free up the channel for reuse */
	spin_lock(&conn->trans->client_lock);
	write_lock_bh(&conn->lock);
	write_lock(&call->state_lock);

	if (conn->channels[call->channel] == call)
		conn->channels[call->channel] = NULL;

	if (conn->out_clientflag && conn->bundle) {
		conn->avail_calls++;
		switch (conn->avail_calls) {
		case 1:
			list_move_tail(&conn->bundle_link,
				       &conn->bundle->avail_conns);
		case 2 ... RXRPC_MAXCALLS - 1:
			ASSERT(conn->channels[0] == NULL ||
			       conn->channels[1] == NULL ||
			       conn->channels[2] == NULL ||
			       conn->channels[3] == NULL);
			break;
		case RXRPC_MAXCALLS:
			list_move_tail(&conn->bundle_link,
				       &conn->bundle->unused_conns);
			ASSERT(conn->channels[0] == NULL &&
			       conn->channels[1] == NULL &&
			       conn->channels[2] == NULL &&
			       conn->channels[3] == NULL);
			break;
		default:
			printk(KERN_ERR "RxRPC: conn->avail_calls=%d\n",
			       conn->avail_calls);
			BUG();
		}
	}

	spin_unlock(&conn->trans->client_lock);

	if (call->state < RXRPC_CALL_COMPLETE &&
	    call->state != RXRPC_CALL_CLIENT_FINAL_ACK) {
		_debug("+++ ABORTING STATE %d +++\n", call->state);
		call->state = RXRPC_CALL_LOCALLY_ABORTED;
		call->abort_code = RX_CALL_DEAD;
		set_bit(RXRPC_CALL_ABORT, &call->events);
		rxrpc_queue_call(call);
	}
	write_unlock(&call->state_lock);
	write_unlock_bh(&conn->lock);

	/* clean up the Rx queue */
	if (!skb_queue_empty(&call->rx_queue) ||
	    !skb_queue_empty(&call->rx_oos_queue)) {
		struct rxrpc_skb_priv *sp;
		struct sk_buff *skb;

		_debug("purge Rx queues");

		spin_lock_bh(&call->lock);
		while ((skb = skb_dequeue(&call->rx_queue)) ||
		       (skb = skb_dequeue(&call->rx_oos_queue))) {
			sp = rxrpc_skb(skb);
			if (sp->call) {
				ASSERTCMP(sp->call, ==, call);
				rxrpc_put_call(call);
				sp->call = NULL;
			}
			skb->destructor = NULL;
			spin_unlock_bh(&call->lock);

			_debug("- zap %s %%%u #%u",
			       rxrpc_pkts[sp->hdr.type],
			       ntohl(sp->hdr.serial),
			       ntohl(sp->hdr.seq));
			rxrpc_free_skb(skb);
			spin_lock_bh(&call->lock);
		}
		spin_unlock_bh(&call->lock);

		ASSERTCMP(call->state, !=, RXRPC_CALL_COMPLETE);
	}

	del_timer_sync(&call->resend_timer);
	del_timer_sync(&call->ack_timer);
	del_timer_sync(&call->lifetimer);
	call->deadspan.expires = jiffies + rxrpc_dead_call_timeout * HZ;
	add_timer(&call->deadspan);

	_leave("");
}

/*
 * handle a dead call being ready for reaping
 */
static void rxrpc_dead_call_expired(unsigned long _call)
{
	struct rxrpc_call *call = (struct rxrpc_call *) _call;

	_enter("{%d}", call->debug_id);

	write_lock_bh(&call->state_lock);
	call->state = RXRPC_CALL_DEAD;
	write_unlock_bh(&call->state_lock);
	rxrpc_put_call(call);
}

/*
 * mark a call as to be released, aborting it if it's still in progress
 * - called with softirqs disabled
 */
static void rxrpc_mark_call_released(struct rxrpc_call *call)
{
	bool sched;

	write_lock(&call->state_lock);
	if (call->state < RXRPC_CALL_DEAD) {
		sched = false;
		if (call->state < RXRPC_CALL_COMPLETE) {
			_debug("abort call %p", call);
			call->state = RXRPC_CALL_LOCALLY_ABORTED;
			call->abort_code = RX_CALL_DEAD;
			if (!test_and_set_bit(RXRPC_CALL_ABORT, &call->events))
				sched = true;
		}
		if (!test_and_set_bit(RXRPC_CALL_RELEASE, &call->events))
			sched = true;
		if (sched)
			rxrpc_queue_call(call);
	}
	write_unlock(&call->state_lock);
}

/*
 * release all the calls associated with a socket
 */
void rxrpc_release_calls_on_socket(struct rxrpc_sock *rx)
{
	struct rxrpc_call *call;
	struct rb_node *p;

	_enter("%p", rx);

	read_lock_bh(&rx->call_lock);

	/* mark all the calls as no longer wanting incoming packets */
	for (p = rb_first(&rx->calls); p; p = rb_next(p)) {
		call = rb_entry(p, struct rxrpc_call, sock_node);
		rxrpc_mark_call_released(call);
	}

	/* kill the not-yet-accepted incoming calls */
	list_for_each_entry(call, &rx->secureq, accept_link) {
		rxrpc_mark_call_released(call);
	}

	list_for_each_entry(call, &rx->acceptq, accept_link) {
		rxrpc_mark_call_released(call);
	}

	read_unlock_bh(&rx->call_lock);
	_leave("");
}

/*
 * release a call
 */
void __rxrpc_put_call(struct rxrpc_call *call)
{
	ASSERT(call != NULL);

	_enter("%p{u=%d}", call, atomic_read(&call->usage));

	ASSERTCMP(atomic_read(&call->usage), >, 0);

	if (atomic_dec_and_test(&call->usage)) {
		_debug("call %d dead", call->debug_id);
		ASSERTCMP(call->state, ==, RXRPC_CALL_DEAD);
		rxrpc_queue_work(&call->destroyer);
	}
	_leave("");
}

/*
 * clean up a call
 */
static void rxrpc_cleanup_call(struct rxrpc_call *call)
{
	_net("DESTROY CALL %d", call->debug_id);

	ASSERT(call->socket);

	memset(&call->sock_node, 0xcd, sizeof(call->sock_node));

	del_timer_sync(&call->lifetimer);
	del_timer_sync(&call->deadspan);
	del_timer_sync(&call->ack_timer);
	del_timer_sync(&call->resend_timer);

	ASSERT(test_bit(RXRPC_CALL_RELEASED, &call->flags));
	ASSERTCMP(call->events, ==, 0);
	if (work_pending(&call->processor)) {
		_debug("defer destroy");
		rxrpc_queue_work(&call->destroyer);
		return;
	}

	if (call->conn) {
		spin_lock(&call->conn->trans->peer->lock);
		list_del(&call->error_link);
		spin_unlock(&call->conn->trans->peer->lock);

		write_lock_bh(&call->conn->lock);
		rb_erase(&call->conn_node, &call->conn->calls);
		write_unlock_bh(&call->conn->lock);
		rxrpc_put_connection(call->conn);
	}

	if (call->acks_window) {
		_debug("kill Tx window %d",
		       CIRC_CNT(call->acks_head, call->acks_tail,
				call->acks_winsz));
		smp_mb();
		while (CIRC_CNT(call->acks_head, call->acks_tail,
				call->acks_winsz) > 0) {
			struct rxrpc_skb_priv *sp;
			unsigned long _skb;

			_skb = call->acks_window[call->acks_tail] & ~1;
			sp = rxrpc_skb((struct sk_buff *) _skb);
			_debug("+++ clear Tx %u", ntohl(sp->hdr.seq));
			rxrpc_free_skb((struct sk_buff *) _skb);
			call->acks_tail =
				(call->acks_tail + 1) & (call->acks_winsz - 1);
		}

		kfree(call->acks_window);
	}

	rxrpc_free_skb(call->tx_pending);

	rxrpc_purge_queue(&call->rx_queue);
	ASSERT(skb_queue_empty(&call->rx_oos_queue));
	sock_put(&call->socket->sk);
	kmem_cache_free(rxrpc_call_jar, call);
}

/*
 * destroy a call
 */
static void rxrpc_destroy_call(struct work_struct *work)
{
	struct rxrpc_call *call =
		container_of(work, struct rxrpc_call, destroyer);

	_enter("%p{%d,%d,%p}",
	       call, atomic_read(&call->usage), call->channel, call->conn);

	ASSERTCMP(call->state, ==, RXRPC_CALL_DEAD);

	write_lock_bh(&rxrpc_call_lock);
	list_del_init(&call->link);
	write_unlock_bh(&rxrpc_call_lock);

	rxrpc_cleanup_call(call);
	_leave("");
}

/*
 * preemptively destroy all the call records from a transport endpoint rather
 * than waiting for them to time out
 */
void __exit rxrpc_destroy_all_calls(void)
{
	struct rxrpc_call *call;

	_enter("");
	write_lock_bh(&rxrpc_call_lock);

	while (!list_empty(&rxrpc_calls)) {
		call = list_entry(rxrpc_calls.next, struct rxrpc_call, link);
		_debug("Zapping call %p", call);

		list_del_init(&call->link);

		switch (atomic_read(&call->usage)) {
		case 0:
			ASSERTCMP(call->state, ==, RXRPC_CALL_DEAD);
			break;
		case 1:
			if (del_timer_sync(&call->deadspan) != 0 &&
			    call->state != RXRPC_CALL_DEAD)
				rxrpc_dead_call_expired((unsigned long) call);
			if (call->state != RXRPC_CALL_DEAD)
				break;
		default:
			printk(KERN_ERR "RXRPC:"
			       " Call %p still in use (%d,%d,%s,%lx,%lx)!\n",
			       call, atomic_read(&call->usage),
			       atomic_read(&call->ackr_not_idle),
			       rxrpc_call_states[call->state],
			       call->flags, call->events);
			if (!skb_queue_empty(&call->rx_queue))
				printk(KERN_ERR"RXRPC: Rx queue occupied\n");
			if (!skb_queue_empty(&call->rx_oos_queue))
				printk(KERN_ERR"RXRPC: OOS queue occupied\n");
			break;
		}

		write_unlock_bh(&rxrpc_call_lock);
		cond_resched();
		write_lock_bh(&rxrpc_call_lock);
	}

	write_unlock_bh(&rxrpc_call_lock);
	_leave("");
}

/*
 * handle call lifetime being exceeded
 */
static void rxrpc_call_life_expired(unsigned long _call)
{
	struct rxrpc_call *call = (struct rxrpc_call *) _call;

	if (call->state >= RXRPC_CALL_COMPLETE)
		return;

	_enter("{%d}", call->debug_id);
	read_lock_bh(&call->state_lock);
	if (call->state < RXRPC_CALL_COMPLETE) {
		set_bit(RXRPC_CALL_LIFE_TIMER, &call->events);
		rxrpc_queue_call(call);
	}
	read_unlock_bh(&call->state_lock);
}

/*
 * handle resend timer expiry
 */
static void rxrpc_resend_time_expired(unsigned long _call)
{
	struct rxrpc_call *call = (struct rxrpc_call *) _call;

	_enter("{%d}", call->debug_id);

	if (call->state >= RXRPC_CALL_COMPLETE)
		return;

	read_lock_bh(&call->state_lock);
	clear_bit(RXRPC_CALL_RUN_RTIMER, &call->flags);
	if (call->state < RXRPC_CALL_COMPLETE &&
	    !test_and_set_bit(RXRPC_CALL_RESEND_TIMER, &call->events))
		rxrpc_queue_call(call);
	read_unlock_bh(&call->state_lock);
}

/*
 * handle ACK timer expiry
 */
static void rxrpc_ack_time_expired(unsigned long _call)
{
	struct rxrpc_call *call = (struct rxrpc_call *) _call;

	_enter("{%d}", call->debug_id);

	if (call->state >= RXRPC_CALL_COMPLETE)
		return;

	read_lock_bh(&call->state_lock);
	if (call->state < RXRPC_CALL_COMPLETE &&
	    !test_and_set_bit(RXRPC_CALL_ACK, &call->events))
		rxrpc_queue_call(call);
	read_unlock_bh(&call->state_lock);
}