Merge tag 'rxrpc-rewrite-20160907-1' of... (9103e04b) · Commits · 戴 / test

include/trace/events/rxrpc.h

+3 −8

Original line number	Diff line number	Diff line
		@@ -17,7 +17,8 @@
		#include <linux/tracepoint.h>

		TRACE_EVENT(rxrpc_call,
		TP_PROTO(struct rxrpc_call *call, int op, int usage, int nskb,
		TP_PROTO(struct rxrpc_call *call, enum rxrpc_call_trace op,
		int usage, int nskb,
		const void where, const void aux),

		TP_ARGS(call, op, usage, nskb, where, aux),
		@@ -42,13 +43,7 @@ TRACE_EVENT(rxrpc_call,

		TP_printk("c=%p %s u=%d s=%d p=%pSR a=%p",
		__entry->call,
		(__entry->op == 0 ? "NWc" :
		__entry->op == 1 ? "NWs" :
		__entry->op == 2 ? "SEE" :
		__entry->op == 3 ? "GET" :
		__entry->op == 4 ? "Gsb" :
		__entry->op == 5 ? "PUT" :
		"Psb"),
		rxrpc_call_traces[__entry->op],
		__entry->usage,
		__entry->nskb,
		__entry->where,

net/rxrpc/af_rxrpc.c

+3 −3

Original line number	Diff line number	Diff line
		@@ -294,9 +294,8 @@ EXPORT_SYMBOL(rxrpc_kernel_begin_call);
		void rxrpc_kernel_end_call(struct socket sock, struct rxrpc_call call)
		{
		_enter("%d{%d}", call->debug_id, atomic_read(&call->usage));
		rxrpc_remove_user_ID(rxrpc_sk(sock->sk), call);
		rxrpc_purge_queue(&call->knlrecv_queue);
		rxrpc_put_call(call);
		rxrpc_release_call(rxrpc_sk(sock->sk), call);
		rxrpc_put_call(call, rxrpc_call_put);
		}
		EXPORT_SYMBOL(rxrpc_kernel_end_call);

		@@ -558,6 +557,7 @@ static int rxrpc_create(struct net net, struct socket sock, int protocol,
		return -ENOMEM;

		sock_init_data(sock, sk);
		sock_set_flag(sk, SOCK_RCU_FREE);
		sk->sk_state = RXRPC_UNBOUND;
		sk->sk_write_space = rxrpc_write_space;
		sk->sk_max_ack_backlog = 0;

net/rxrpc/ar-internal.h

+27 −14

Original line number	Diff line number	Diff line
		@@ -35,8 +35,6 @@ struct rxrpc_crypt {
		#define rxrpc_queue_delayed_work(WS,D) \
		queue_delayed_work(rxrpc_workqueue, (WS), (D))

		#define rxrpc_queue_call(CALL) rxrpc_queue_work(&(CALL)->processor)

		struct rxrpc_connection;

		/*
		@@ -397,7 +395,6 @@ enum rxrpc_call_event {
		RXRPC_CALL_EV_ACCEPTED, /* incoming call accepted by userspace app */
		RXRPC_CALL_EV_SECURED, /* incoming call's connection is now secure */
		RXRPC_CALL_EV_POST_ACCEPT, /* need to post an "accept?" message to the app */
		RXRPC_CALL_EV_RELEASE, /* need to release the call's resources */
		};

		/*
		@@ -417,7 +414,6 @@ enum rxrpc_call_state {
		RXRPC_CALL_SERVER_SEND_REPLY, /* - server sending reply */
		RXRPC_CALL_SERVER_AWAIT_ACK, /* - server awaiting final ACK */
		RXRPC_CALL_COMPLETE, /* - call complete */
		RXRPC_CALL_DEAD, /* - call is dead */
		NR__RXRPC_CALL_STATES
		};

		@@ -442,12 +438,10 @@ struct rxrpc_call {
		struct rcu_head rcu;
		struct rxrpc_connection conn; / connection carrying call */
		struct rxrpc_peer peer; / Peer record for remote address */
		struct rxrpc_sock socket; / socket responsible */
		struct rxrpc_sock __rcu socket; / socket responsible */
		struct timer_list lifetimer; /* lifetime remaining on call */
		struct timer_list deadspan; /* reap timer for re-ACK'ing, etc */
		struct timer_list ack_timer; /* ACK generation timer */
		struct timer_list resend_timer; /* Tx resend timer */
		struct work_struct destroyer; /* call destroyer */
		struct work_struct processor; /* packet processor and ACK generator */
		rxrpc_notify_rx_t notify_rx; /* kernel service Rx notification function */
		struct list_head link; /* link in master call list */
		@@ -475,6 +469,7 @@ struct rxrpc_call {
		atomic_t skb_count; /* Outstanding packets on this call */
		atomic_t sequence; /* Tx data packet sequence counter */
		u16 service_id; /* service ID */
		u8 security_ix; /* Security type */
		u32 call_id; /* call ID on connection */
		u32 cid; /* connection ID plus channel index */
		int debug_id; /* debug ID for printks */
		@@ -508,6 +503,24 @@ struct rxrpc_call {
		unsigned long ackr_window[RXRPC_ACKR_WINDOW_ASZ + 1];
		};

		enum rxrpc_call_trace {
		rxrpc_call_new_client,
		rxrpc_call_new_service,
		rxrpc_call_queued,
		rxrpc_call_queued_ref,
		rxrpc_call_seen,
		rxrpc_call_got,
		rxrpc_call_got_skb,
		rxrpc_call_got_userid,
		rxrpc_call_put,
		rxrpc_call_put_skb,
		rxrpc_call_put_userid,
		rxrpc_call_put_noqueue,
		rxrpc_call__nr_trace
		};

		extern const char rxrpc_call_traces[rxrpc_call__nr_trace][4];

		#include <trace/events/rxrpc.h>

		/*
		@@ -539,7 +552,6 @@ void rxrpc_process_call(struct work_struct *);
		extern const char *const rxrpc_call_states[];
		extern const char *const rxrpc_call_completions[];
		extern unsigned int rxrpc_max_call_lifetime;
		extern unsigned int rxrpc_dead_call_expiry;
		extern struct kmem_cache *rxrpc_call_jar;
		extern struct list_head rxrpc_calls;
		extern rwlock_t rxrpc_call_lock;
		@@ -552,11 +564,13 @@ struct rxrpc_call rxrpc_new_client_call(struct rxrpc_sock ,
		struct rxrpc_call rxrpc_incoming_call(struct rxrpc_sock ,
		struct rxrpc_connection *,
		struct sk_buff *);
		void rxrpc_release_call(struct rxrpc_call *);
		void rxrpc_release_call(struct rxrpc_sock , struct rxrpc_call );
		void rxrpc_release_calls_on_socket(struct rxrpc_sock *);
		bool __rxrpc_queue_call(struct rxrpc_call *);
		bool rxrpc_queue_call(struct rxrpc_call *);
		void rxrpc_see_call(struct rxrpc_call *);
		void rxrpc_get_call(struct rxrpc_call *);
		void rxrpc_put_call(struct rxrpc_call *);
		void rxrpc_get_call(struct rxrpc_call *, enum rxrpc_call_trace);
		void rxrpc_put_call(struct rxrpc_call *, enum rxrpc_call_trace);
		void rxrpc_get_call_for_skb(struct rxrpc_call , struct sk_buff );
		void rxrpc_put_call_for_skb(struct rxrpc_call , struct sk_buff );
		void __exit rxrpc_destroy_all_calls(void);
		@@ -720,7 +734,6 @@ static inline void rxrpc_put_connection(struct rxrpc_connection *conn)
		}
		}


		static inline bool rxrpc_queue_conn(struct rxrpc_connection *conn)
		{
		if (!rxrpc_get_connection_maybe(conn))
		@@ -817,6 +830,7 @@ extern const char *rxrpc_acks(u8 reason);
		/*
		* output.c
		*/
		int rxrpc_send_call_packet(struct rxrpc_call *, u8);
		int rxrpc_send_data_packet(struct rxrpc_connection , struct sk_buff );

		/*
		@@ -862,7 +876,6 @@ extern const struct file_operations rxrpc_connection_seq_fops;
		/*
		* recvmsg.c
		*/
		void rxrpc_remove_user_ID(struct rxrpc_sock , struct rxrpc_call );
		int rxrpc_recvmsg(struct socket , struct msghdr , size_t, int);

		/*

net/rxrpc/call_accept.c

+19 −44

Original line number	Diff line number	Diff line
		@@ -115,7 +115,7 @@ static int rxrpc_accept_incoming_call(struct rxrpc_local *local,

		write_lock(&rx->call_lock);
		if (!test_and_set_bit(RXRPC_CALL_INIT_ACCEPT, &call->flags)) {
		rxrpc_get_call(call);
		rxrpc_get_call(call, rxrpc_call_got);

		spin_lock(&call->conn->state_lock);
		if (sp->hdr.securityIndex > 0 &&
		@@ -155,7 +155,7 @@ static int rxrpc_accept_incoming_call(struct rxrpc_local *local,
		_debug("done");
		read_unlock_bh(&local->services_lock);
		rxrpc_free_skb(notification);
		rxrpc_put_call(call);
		rxrpc_put_call(call, rxrpc_call_put);
		_leave(" = 0");
		return 0;

		@@ -163,14 +163,8 @@ invalid_service:
		_debug("invalid");
		read_unlock_bh(&local->services_lock);

		read_lock_bh(&call->state_lock);
		if (!test_bit(RXRPC_CALL_RELEASED, &call->flags) &&
		!test_and_set_bit(RXRPC_CALL_EV_RELEASE, &call->events)) {
		rxrpc_get_call(call);
		rxrpc_queue_call(call);
		}
		read_unlock_bh(&call->state_lock);
		rxrpc_put_call(call);
		rxrpc_release_call(rx, call);
		rxrpc_put_call(call, rxrpc_call_put);
		ret = -ECONNREFUSED;
		error:
		rxrpc_free_skb(notification);
		@@ -236,13 +230,11 @@ found_service:
		if (sk_acceptq_is_full(&rx->sk))
		goto backlog_full;
		sk_acceptq_added(&rx->sk);
		sock_hold(&rx->sk);
		read_unlock_bh(&local->services_lock);

		ret = rxrpc_accept_incoming_call(local, rx, skb, &srx);
		if (ret < 0)
		sk_acceptq_removed(&rx->sk);
		sock_put(&rx->sk);
		switch (ret) {
		case -ECONNRESET: /* old calls are ignored */
		case -ECONNABORTED: /* aborted calls are reaborted or ignored */
		@@ -333,14 +325,12 @@ struct rxrpc_call rxrpc_accept_call(struct rxrpc_sock rx,
		case RXRPC_CALL_COMPLETE:
		ret = call->error;
		goto out_release;
		case RXRPC_CALL_DEAD:
		ret = -ETIME;
		goto out_discard;
		default:
		BUG();
		}

		/* formalise the acceptance */
		rxrpc_get_call(call, rxrpc_call_got_userid);
		call->notify_rx = notify_rx;
		call->user_call_ID = user_call_ID;
		rb_link_node(&call->sock_node, parent, pp);
		@@ -349,25 +339,20 @@ struct rxrpc_call rxrpc_accept_call(struct rxrpc_sock rx,
		BUG();
		if (test_and_set_bit(RXRPC_CALL_EV_ACCEPTED, &call->events))
		BUG();
		rxrpc_queue_call(call);

		rxrpc_get_call(call);
		write_unlock_bh(&call->state_lock);
		write_unlock(&rx->call_lock);
		rxrpc_queue_call(call);
		_leave(" = %p{%d}", call, call->debug_id);
		return call;

		/* if the call is already dying or dead, then we leave the socket's ref
		* on it to be released by rxrpc_dead_call_expired() as induced by
		* rxrpc_release_call() */
		out_release:
		_debug("release %p", call);
		if (!test_bit(RXRPC_CALL_RELEASED, &call->flags) &&
		!test_and_set_bit(RXRPC_CALL_EV_RELEASE, &call->events))
		rxrpc_queue_call(call);
		out_discard:
		write_unlock_bh(&call->state_lock);
		_debug("discard %p", call);
		write_unlock(&rx->call_lock);
		_debug("release %p", call);
		rxrpc_release_call(rx, call);
		_leave(" = %d", ret);
		return ERR_PTR(ret);
		out:
		write_unlock(&rx->call_lock);
		_leave(" = %d", ret);
		@@ -390,8 +375,11 @@ int rxrpc_reject_call(struct rxrpc_sock *rx)
		write_lock(&rx->call_lock);

		ret = -ENODATA;
		if (list_empty(&rx->acceptq))
		goto out;
		if (list_empty(&rx->acceptq)) {
		write_unlock(&rx->call_lock);
		_leave(" = -ENODATA");
		return -ENODATA;
		}

		/* dequeue the first call and check it's still valid */
		call = list_entry(rx->acceptq.next, struct rxrpc_call, accept_link);
		@@ -407,30 +395,17 @@ int rxrpc_reject_call(struct rxrpc_sock *rx)
		if (test_and_set_bit(RXRPC_CALL_EV_REJECT_BUSY, &call->events))
		rxrpc_queue_call(call);
		ret = 0;
		goto out_release;
		break;
		case RXRPC_CALL_COMPLETE:
		ret = call->error;
		goto out_release;
		case RXRPC_CALL_DEAD:
		ret = -ETIME;
		goto out_discard;
		break;
		default:
		BUG();
		}

		/* if the call is already dying or dead, then we leave the socket's ref
		* on it to be released by rxrpc_dead_call_expired() as induced by
		* rxrpc_release_call() */
		out_release:
		_debug("release %p", call);
		if (!test_bit(RXRPC_CALL_RELEASED, &call->flags) &&
		!test_and_set_bit(RXRPC_CALL_EV_RELEASE, &call->events))
		rxrpc_queue_call(call);
		out_discard:
		write_unlock_bh(&call->state_lock);
		_debug("discard %p", call);
		out:
		write_unlock(&rx->call_lock);
		rxrpc_release_call(rx, call);
		_leave(" = %d", ret);
		return ret;
		}

net/rxrpc/call_event.c

+30 −44

Original line number	Diff line number	Diff line
		@@ -811,8 +811,9 @@ static int rxrpc_post_message(struct rxrpc_call *call, u32 mark, u32 error,
		}

		/*
		* handle background processing of incoming call packets and ACK / abort
		* generation
		* Handle background processing of incoming call packets and ACK / abort
		* generation. A ref on the call is donated to us by whoever queued the work
		* item.
		*/
		void rxrpc_process_call(struct work_struct *work)
		{
		@@ -827,6 +828,7 @@ void rxrpc_process_call(struct work_struct *work)
		unsigned long bits;
		__be32 data, pad;
		size_t len;
		bool requeue = false;
		int loop, nbit, ioc, ret, mtu;
		u32 serial, abort_code = RX_PROTOCOL_ERROR;
		u8 *acks = NULL;
		@@ -838,6 +840,11 @@ void rxrpc_process_call(struct work_struct *work)
		call->debug_id, rxrpc_call_states[call->state], call->events,
		(jiffies - call->creation_jif) / (HZ / 10));

		if (call->state >= RXRPC_CALL_COMPLETE) {
		rxrpc_put_call(call, rxrpc_call_put);
		return;
		}

		if (!call->conn)
		goto skip_msg_init;

		@@ -1088,16 +1095,21 @@ skip_msg_init:
		spin_lock_bh(&call->lock);

		if (call->state == RXRPC_CALL_SERVER_SECURING) {
		struct rxrpc_sock *rx;
		_debug("securing");
		write_lock(&call->socket->call_lock);
		if (!test_bit(RXRPC_CALL_RELEASED, &call->flags) &&
		!test_bit(RXRPC_CALL_EV_RELEASE, &call->events)) {
		rcu_read_lock();
		rx = rcu_dereference(call->socket);
		if (rx) {
		write_lock(&rx->call_lock);
		if (!test_bit(RXRPC_CALL_RELEASED, &call->flags)) {
		_debug("not released");
		call->state = RXRPC_CALL_SERVER_ACCEPTING;
		list_move_tail(&call->accept_link,
		&call->socket->acceptq);
		&rx->acceptq);
		}
		write_unlock(&call->socket->call_lock);
		write_unlock(&rx->call_lock);
		}
		rcu_read_unlock();
		read_lock(&call->state_lock);
		if (call->state < RXRPC_CALL_COMPLETE)
		set_bit(RXRPC_CALL_EV_POST_ACCEPT, &call->events);
		@@ -1139,11 +1151,6 @@ skip_msg_init:
		goto maybe_reschedule;
		}

		if (test_bit(RXRPC_CALL_EV_RELEASE, &call->events)) {
		rxrpc_release_call(call);
		clear_bit(RXRPC_CALL_EV_RELEASE, &call->events);
		}

		/* other events may have been raised since we started checking */
		goto maybe_reschedule;

		@@ -1209,10 +1216,8 @@ send_message_2:
		&msg, iov, ioc, len);
		if (ret < 0) {
		_debug("sendmsg failed: %d", ret);
		read_lock_bh(&call->state_lock);
		if (call->state < RXRPC_CALL_DEAD)
		rxrpc_queue_call(call);
		read_unlock_bh(&call->state_lock);
		if (call->state < RXRPC_CALL_COMPLETE)
		requeue = true;
		goto error;
		}

		@@ -1245,41 +1250,22 @@ send_message_2:

		kill_ACKs:
		del_timer_sync(&call->ack_timer);
		if (test_and_clear_bit(RXRPC_CALL_EV_ACK_FINAL, &call->events))
		rxrpc_put_call(call);
		clear_bit(RXRPC_CALL_EV_ACK, &call->events);

		maybe_reschedule:
		if (call->events \|\| !skb_queue_empty(&call->rx_queue)) {
		read_lock_bh(&call->state_lock);
		if (call->state < RXRPC_CALL_DEAD)
		rxrpc_queue_call(call);
		read_unlock_bh(&call->state_lock);
		}

		/* don't leave aborted connections on the accept queue */
		if (call->state >= RXRPC_CALL_COMPLETE &&
		!list_empty(&call->accept_link)) {
		_debug("X unlinking once-pending call %p { e=%lx f=%lx c=%x }",
		call, call->events, call->flags, call->conn->proto.cid);

		read_lock_bh(&call->state_lock);
		if (!test_bit(RXRPC_CALL_RELEASED, &call->flags) &&
		!test_and_set_bit(RXRPC_CALL_EV_RELEASE, &call->events))
		rxrpc_queue_call(call);
		read_unlock_bh(&call->state_lock);
		if (call->state < RXRPC_CALL_COMPLETE)
		requeue = true;
		}

		error:
		kfree(acks);

		/* because we don't want two CPUs both processing the work item for one
		* call at the same time, we use a flag to note when it's busy; however
		* this means there's a race between clearing the flag and setting the
		* work pending bit and the work item being processed again */
		if (call->events && !work_pending(&call->processor)) {
		if ((requeue \|\| call->events) && !work_pending(&call->processor)) {
		_debug("jumpstart %x", call->conn->proto.cid);
		rxrpc_queue_call(call);
		__rxrpc_queue_call(call);
		} else {
		rxrpc_put_call(call, rxrpc_call_put);
		}

		_leave("");

Admin message