ar-connection.c

linux 内核源代码

		}

		if (!list_empty(&bundle->unused_conns)) {
			_debug("unused");
			conn = list_entry(bundle->unused_conns.next,
					  struct rxrpc_connection,
					  bundle_link);
			ASSERTCMP(conn->avail_calls, ==, RXRPC_MAXCALLS);
			conn->avail_calls = RXRPC_MAXCALLS - 1;
			ASSERT(conn->channels[0] == NULL &&
			       conn->channels[1] == NULL &&
			       conn->channels[2] == NULL &&
			       conn->channels[3] == NULL);
			atomic_inc(&conn->usage);
			list_move(&conn->bundle_link, &bundle->avail_conns);
			break;
		}

		/* need to allocate a new connection */
		_debug("get new conn [%d]", bundle->num_conns);

		spin_unlock(&trans->client_lock);

		if (signal_pending(current))
			goto interrupted;

		if (bundle->num_conns >= 20) {
			_debug("too many conns");

			if (!(gfp & __GFP_WAIT)) {
				_leave(" = -EAGAIN");
				return -EAGAIN;
			}

			add_wait_queue(&bundle->chanwait, &myself);
			for (;;) {
				set_current_state(TASK_INTERRUPTIBLE);
				if (bundle->num_conns < 20 ||
				    !list_empty(&bundle->unused_conns) ||
				    !list_empty(&bundle->avail_conns))
					break;
				if (signal_pending(current))
					goto interrupted_dequeue;
				schedule();
			}
			remove_wait_queue(&bundle->chanwait, &myself);
			__set_current_state(TASK_RUNNING);
			spin_lock(&trans->client_lock);
			continue;
		}

		/* not yet present - create a candidate for a new connection and then
		 * redo the check */
		candidate = rxrpc_alloc_connection(gfp);
		if (IS_ERR(candidate)) {
			_leave(" = %ld", PTR_ERR(candidate));
			return PTR_ERR(candidate);
		}

		candidate->trans = trans;
		candidate->bundle = bundle;
		candidate->service_id = bundle->service_id;
		candidate->epoch = rxrpc_epoch;
		candidate->in_clientflag = 0;
		candidate->out_clientflag = RXRPC_CLIENT_INITIATED;
		candidate->cid = 0;
		candidate->state = RXRPC_CONN_CLIENT;
		candidate->avail_calls = RXRPC_MAXCALLS;
		candidate->security_level = rx->min_sec_level;
		candidate->key = key_get(bundle->key);

		ret = rxrpc_init_client_conn_security(candidate);
		if (ret < 0) {
			key_put(candidate->key);
			kfree(candidate);
			_leave(" = %d [key]", ret);
			return ret;
		}

		write_lock_bh(&rxrpc_connection_lock);
		list_add_tail(&candidate->link, &rxrpc_connections);
		write_unlock_bh(&rxrpc_connection_lock);

		spin_lock(&trans->client_lock);

		list_add(&candidate->bundle_link, &bundle->unused_conns);
		bundle->num_conns++;
		atomic_inc(&bundle->usage);
		atomic_inc(&trans->usage);

		_net("CONNECT new %d on TRANS %d",
		     candidate->debug_id, candidate->trans->debug_id);

		rxrpc_assign_connection_id(candidate);
		if (candidate->security)
			candidate->security->prime_packet_security(candidate);

		/* leave the candidate lurking in zombie mode attached to the
		 * bundle until we're ready for it */
		rxrpc_put_connection(candidate);
		candidate = NULL;
	}

	/* we've got a connection with a free channel and we can now attach the
	 * call to it
	 * - we're holding the transport's client lock
	 * - we're holding a reference on the connection
	 * - we're holding a reference on the bundle
	 */
	for (chan = 0; chan < RXRPC_MAXCALLS; chan++)
		if (!conn->channels[chan])
			goto found_channel;
	ASSERT(conn->channels[0] == NULL ||
	       conn->channels[1] == NULL ||
	       conn->channels[2] == NULL ||
	       conn->channels[3] == NULL);
	BUG();

found_channel:
	conn->channels[chan] = call;
	call->conn = conn;
	call->channel = chan;
	call->cid = conn->cid | htonl(chan);
	call->call_id = htonl(++conn->call_counter);

	_net("CONNECT client on conn %d chan %d as call %x",
	     conn->debug_id, chan, ntohl(call->call_id));

	ASSERTCMP(conn->avail_calls, <, RXRPC_MAXCALLS);
	spin_unlock(&trans->client_lock);

	rxrpc_add_call_ID_to_conn(conn, call);

	_leave(" = 0");
	return 0;

interrupted_dequeue:
	remove_wait_queue(&bundle->chanwait, &myself);
	__set_current_state(TASK_RUNNING);
interrupted:
	_leave(" = -ERESTARTSYS");
	return -ERESTARTSYS;
}

/*
 * get a record of an incoming connection
 */
struct rxrpc_connection *
rxrpc_incoming_connection(struct rxrpc_transport *trans,
			  struct rxrpc_header *hdr,
			  gfp_t gfp)
{
	struct rxrpc_connection *conn, *candidate = NULL;
	struct rb_node *p, **pp;
	const char *new = "old";
	__be32 epoch;
	u32 conn_id;

	_enter("");

	ASSERT(hdr->flags & RXRPC_CLIENT_INITIATED);

	epoch = hdr->epoch;
	conn_id = ntohl(hdr->cid) & RXRPC_CIDMASK;

	/* search the connection list first */
	read_lock_bh(&trans->conn_lock);

	p = trans->server_conns.rb_node;
	while (p) {
		conn = rb_entry(p, struct rxrpc_connection, node);

		_debug("maybe %x", conn->real_conn_id);

		if (epoch < conn->epoch)
			p = p->rb_left;
		else if (epoch > conn->epoch)
			p = p->rb_right;
		else if (conn_id < conn->real_conn_id)
			p = p->rb_left;
		else if (conn_id > conn->real_conn_id)
			p = p->rb_right;
		else
			goto found_extant_connection;
	}
	read_unlock_bh(&trans->conn_lock);

	/* not yet present - create a candidate for a new record and then
	 * redo the search */
	candidate = rxrpc_alloc_connection(gfp);
	if (!candidate) {
		_leave(" = -ENOMEM");
		return ERR_PTR(-ENOMEM);
	}

	candidate->trans = trans;
	candidate->epoch = hdr->epoch;
	candidate->cid = hdr->cid & __constant_cpu_to_be32(RXRPC_CIDMASK);
	candidate->service_id = hdr->serviceId;
	candidate->security_ix = hdr->securityIndex;
	candidate->in_clientflag = RXRPC_CLIENT_INITIATED;
	candidate->out_clientflag = 0;
	candidate->real_conn_id = conn_id;
	candidate->state = RXRPC_CONN_SERVER;
	if (candidate->service_id)
		candidate->state = RXRPC_CONN_SERVER_UNSECURED;

	write_lock_bh(&trans->conn_lock);

	pp = &trans->server_conns.rb_node;
	p = NULL;
	while (*pp) {
		p = *pp;
		conn = rb_entry(p, struct rxrpc_connection, node);

		if (epoch < conn->epoch)
			pp = &(*pp)->rb_left;
		else if (epoch > conn->epoch)
			pp = &(*pp)->rb_right;
		else if (conn_id < conn->real_conn_id)
			pp = &(*pp)->rb_left;
		else if (conn_id > conn->real_conn_id)
			pp = &(*pp)->rb_right;
		else
			goto found_extant_second;
	}

	/* we can now add the new candidate to the list */
	conn = candidate;
	candidate = NULL;
	rb_link_node(&conn->node, p, pp);
	rb_insert_color(&conn->node, &trans->server_conns);
	atomic_inc(&conn->trans->usage);

	write_unlock_bh(&trans->conn_lock);

	write_lock_bh(&rxrpc_connection_lock);
	list_add_tail(&conn->link, &rxrpc_connections);
	write_unlock_bh(&rxrpc_connection_lock);

	new = "new";

success:
	_net("CONNECTION %s %d {%x}", new, conn->debug_id, conn->real_conn_id);

	_leave(" = %p {u=%d}", conn, atomic_read(&conn->usage));
	return conn;

	/* we found the connection in the list immediately */
found_extant_connection:
	if (hdr->securityIndex != conn->security_ix) {
		read_unlock_bh(&trans->conn_lock);
		goto security_mismatch;
	}
	atomic_inc(&conn->usage);
	read_unlock_bh(&trans->conn_lock);
	goto success;

	/* we found the connection on the second time through the list */
found_extant_second:
	if (hdr->securityIndex != conn->security_ix) {
		write_unlock_bh(&trans->conn_lock);
		goto security_mismatch;
	}
	atomic_inc(&conn->usage);
	write_unlock_bh(&trans->conn_lock);
	kfree(candidate);
	goto success;

security_mismatch:
	kfree(candidate);
	_leave(" = -EKEYREJECTED");
	return ERR_PTR(-EKEYREJECTED);
}

/*
 * find a connection based on transport and RxRPC connection ID for an incoming
 * packet
 */
struct rxrpc_connection *rxrpc_find_connection(struct rxrpc_transport *trans,
					       struct rxrpc_header *hdr)
{
	struct rxrpc_connection *conn;
	struct rb_node *p;
	__be32 epoch;
	u32 conn_id;

	_enter(",{%x,%x}", ntohl(hdr->cid), hdr->flags);

	read_lock_bh(&trans->conn_lock);

	conn_id = ntohl(hdr->cid) & RXRPC_CIDMASK;
	epoch = hdr->epoch;

	if (hdr->flags & RXRPC_CLIENT_INITIATED)
		p = trans->server_conns.rb_node;
	else
		p = trans->client_conns.rb_node;

	while (p) {
		conn = rb_entry(p, struct rxrpc_connection, node);

		_debug("maybe %x", conn->real_conn_id);

		if (epoch < conn->epoch)
			p = p->rb_left;
		else if (epoch > conn->epoch)
			p = p->rb_right;
		else if (conn_id < conn->real_conn_id)
			p = p->rb_left;
		else if (conn_id > conn->real_conn_id)
			p = p->rb_right;
		else
			goto found;
	}

	read_unlock_bh(&trans->conn_lock);
	_leave(" = NULL");
	return NULL;

found:
	atomic_inc(&conn->usage);
	read_unlock_bh(&trans->conn_lock);
	_leave(" = %p", conn);
	return conn;
}

/*
 * release a virtual connection
 */
void rxrpc_put_connection(struct rxrpc_connection *conn)
{
	_enter("%p{u=%d,d=%d}",
	       conn, atomic_read(&conn->usage), conn->debug_id);

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

	conn->put_time = get_seconds();
	if (atomic_dec_and_test(&conn->usage)) {
		_debug("zombie");
		rxrpc_queue_delayed_work(&rxrpc_connection_reap, 0);
	}

	_leave("");
}

/*
 * destroy a virtual connection
 */
static void rxrpc_destroy_connection(struct rxrpc_connection *conn)
{
	_enter("%p{%d}", conn, atomic_read(&conn->usage));

	ASSERTCMP(atomic_read(&conn->usage), ==, 0);

	_net("DESTROY CONN %d", conn->debug_id);

	if (conn->bundle)
		rxrpc_put_bundle(conn->trans, conn->bundle);

	ASSERT(RB_EMPTY_ROOT(&conn->calls));
	rxrpc_purge_queue(&conn->rx_queue);

	rxrpc_clear_conn_security(conn);
	rxrpc_put_transport(conn->trans);
	kfree(conn);
	_leave("");
}

/*
 * reap dead connections
 */
void rxrpc_connection_reaper(struct work_struct *work)
{
	struct rxrpc_connection *conn, *_p;
	unsigned long now, earliest, reap_time;

	LIST_HEAD(graveyard);

	_enter("");

	now = get_seconds();
	earliest = ULONG_MAX;

	write_lock_bh(&rxrpc_connection_lock);
	list_for_each_entry_safe(conn, _p, &rxrpc_connections, link) {
		_debug("reap CONN %d { u=%d,t=%ld }",
		       conn->debug_id, atomic_read(&conn->usage),
		       (long) now - (long) conn->put_time);

		if (likely(atomic_read(&conn->usage) > 0))
			continue;

		spin_lock(&conn->trans->client_lock);
		write_lock(&conn->trans->conn_lock);
		reap_time = conn->put_time + rxrpc_connection_timeout;

		if (atomic_read(&conn->usage) > 0) {
			;
		} else if (reap_time <= now) {
			list_move_tail(&conn->link, &graveyard);
			if (conn->out_clientflag)
				rb_erase(&conn->node,
					 &conn->trans->client_conns);
			else
				rb_erase(&conn->node,
					 &conn->trans->server_conns);
			if (conn->bundle) {
				list_del_init(&conn->bundle_link);
				conn->bundle->num_conns--;
			}

		} else if (reap_time < earliest) {
			earliest = reap_time;
		}

		write_unlock(&conn->trans->conn_lock);
		spin_unlock(&conn->trans->client_lock);
	}
	write_unlock_bh(&rxrpc_connection_lock);

	if (earliest != ULONG_MAX) {
		_debug("reschedule reaper %ld", (long) earliest - now);
		ASSERTCMP(earliest, >, now);
		rxrpc_queue_delayed_work(&rxrpc_connection_reap,
					 (earliest - now) * HZ);
	}

	/* then destroy all those pulled out */
	while (!list_empty(&graveyard)) {
		conn = list_entry(graveyard.next, struct rxrpc_connection,
				  link);
		list_del_init(&conn->link);

		ASSERTCMP(atomic_read(&conn->usage), ==, 0);
		rxrpc_destroy_connection(conn);
	}

	_leave("");
}

/*
 * preemptively destroy all the connection records rather than waiting for them
 * to time out
 */
void __exit rxrpc_destroy_all_connections(void)
{
	_enter("");

	rxrpc_connection_timeout = 0;
	cancel_delayed_work(&rxrpc_connection_reap);
	rxrpc_queue_delayed_work(&rxrpc_connection_reap, 0);

	_leave("");
}