From patchwork Mon Apr 8 23:48:34 2013 Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit X-Patchwork-Submitter: Alex Elder X-Patchwork-Id: 2412411 Return-Path: X-Original-To: patchwork-ceph-devel@patchwork.kernel.org Delivered-To: patchwork-process-083081@patchwork1.kernel.org Received: from vger.kernel.org (vger.kernel.org [209.132.180.67]) by patchwork1.kernel.org (Postfix) with ESMTP id AEAD23FD1A for ; Mon, 8 Apr 2013 23:48:39 +0000 (UTC) Received: (majordomo@vger.kernel.org) by vger.kernel.org via listexpand id S936515Ab3DHXsi (ORCPT ); Mon, 8 Apr 2013 19:48:38 -0400 Received: from mail-oa0-f47.google.com ([209.85.219.47]:59117 "EHLO mail-oa0-f47.google.com" rhost-flags-OK-OK-OK-OK) by vger.kernel.org with ESMTP id S1761645Ab3DHXsi (ORCPT ); Mon, 8 Apr 2013 19:48:38 -0400 Received: by mail-oa0-f47.google.com with SMTP id o17so6746208oag.20 for ; Mon, 08 Apr 2013 16:48:37 -0700 (PDT) X-Google-DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=google.com; s=20120113; h=x-received:message-id:date:from:user-agent:mime-version:to:cc :subject:references:in-reply-to:content-type :content-transfer-encoding:x-gm-message-state; bh=Zsxbh1Omro79E5Y0R8rb9uAdYuBG4oNP+2z+4cFqbM4=; b=D2QIMT4La1hUzqZw4J7TLLG9h7jkmpeL5p3sF55UsSkSCtYc1qb1mT5DjINoethQjv 2zvMpxo1sSqEm7Chcqt4W69h6G9p0F0W0Tamu3UdLvkA2McqGukx3YIbuGPc5lHLLFUn iO/bboWA3r6hBm8ghmDqlP/IyMId4zOFf3pkD1WTTUKkdSCi4OQO/q7TlNISNhdNGn96 KRx3UGfHWJ35bAsygMb7/2vRyqIgi18WnRVWc2bQQQZG1F9Gv1JqpX6gBSC61qvNqJN6 tJXHrhp5I9+ya945Owcua/2A1R1AuM4Py+gaiGQCgPu3l3teQbwSqWj1FTSGth4+LeKh W5lw== X-Received: by 10.60.142.103 with SMTP id rv7mr16724435oeb.34.1365464917438; Mon, 08 Apr 2013 16:48:37 -0700 (PDT) Received: from [172.22.22.4] (c-71-195-31-37.hsd1.mn.comcast.net. [71.195.31.37]) by mx.google.com with ESMTPS id ri7sm27840276oeb.6.2013.04.08.16.48.35 (version=TLSv1 cipher=ECDHE-RSA-RC4-SHA bits=128/128); Mon, 08 Apr 2013 16:48:36 -0700 (PDT) Message-ID: <51635752.5080507@inktank.com> Date: Mon, 08 Apr 2013 18:48:34 -0500 From: Alex Elder User-Agent: Mozilla/5.0 (X11; Linux x86_64; rv:17.0) Gecko/20130308 Thunderbird/17.0.4 MIME-Version: 1.0 To: "ceph-devel@vger.kernel.org" CC: Josh Durgin Subject: [PATCH 5/6, v3] libceph: implement multiple data items in a message References: <515F4F01.2000704@inktank.com> <515F4F96.4010308@inktank.com> <515F6009.3050704@inktank.com> In-Reply-To: <515F6009.3050704@inktank.com> X-Gm-Message-State: ALoCoQkYMH3dlLvGygb3zxA5icNwbm2W7kKwx5bOgzVVrmgllgacg0utTxuG1c2o2DdlR9x6LKU0 Sender: ceph-devel-owner@vger.kernel.org Precedence: bulk List-ID: X-Mailing-List: ceph-devel@vger.kernel.org Sorry, I've found another problem with this patch. I've updated it below, and have updated the patches that followed it. The result is available in the "review/wip-3761-5" branch of the ceph-client git repository. Previously there were assertions that caught attempts to add more than one data item to a message. Those need to go away, now that we support multiple data items. This patch adds support to the messenger for more than one data item in its data list. A message data cursor has two more fields to support this: - a count of the number of bytes left to be consumed across all data items in the list, "total_resid" - a pointer to the head of the list (for validation only) The cursor initialization routine has been split into two parts: the outer one, which initializes the cursor for traversing the entire list of data items; and the inner one, which initializes the cursor to start processing a single data item. When a message cursor is first initialized, the outer initialization routine sets total_resid to the length provided. The data pointer is initialized to the first data item on the list. From there, the inner initialization routine finishes by setting up to process the data item the cursor points to. Advancing the cursor consumes bytes in total_resid. If the resid field reaches zero, it means the current data item is fully consumed. If total_resid indicates there is more data, the cursor is advanced to point to the next data item, and then the inner initialization routine prepares for using that. (A check is made at this point to make sure we don't wrap around the front of the list.) The type-specific init routines are modified so they can be given a length that's larger than what the data item can support. The resid field is initialized to the smaller of the provided length and the length of the entire data item. When total_resid reaches zero, we're done. This resolves: http://tracker.ceph.com/issues/3761 Signed-off-by: Alex Elder Reviewed-by: Josh Durgin --- v2: set total_resid to the passed-in length when initializing cursor v3: turn off single data item assertions; it's ok now include/linux/ceph/messenger.h | 5 ++++- net/ceph/messenger.c | 48 ++++++++++++++++++++++++++-------------- 2 files changed, 36 insertions(+), 17 deletions(-) cursor->vector_offset = 0; @@ -833,9 +833,8 @@ static void ceph_msg_data_pages_cursor_init(struct ceph_msg_data_cursor *cursor, BUG_ON(!data->pages); BUG_ON(!data->length); - BUG_ON(length > data->length); /* short reads are OK */ - cursor->resid = length; + cursor->resid = min(length, data->length); page_count = calc_pages_for(data->alignment, (u64)data->length); cursor->page_offset = data->alignment & ~PAGE_MASK; cursor->page_index = 0; @@ -904,7 +903,6 @@ ceph_msg_data_pagelist_cursor_init(struct ceph_msg_data_cursor *cursor, pagelist = data->pagelist; BUG_ON(!pagelist); - BUG_ON(length > pagelist->length); /* short reads are OK */ if (!length) return; /* pagelist can be assigned but empty */ @@ -912,7 +910,7 @@ ceph_msg_data_pagelist_cursor_init(struct ceph_msg_data_cursor *cursor, BUG_ON(list_empty(&pagelist->head)); page = list_first_entry(&pagelist->head, struct page, lru); - cursor->resid = length; + cursor->resid = min(length, pagelist->length); cursor->page = page; cursor->offset = 0; cursor->last_piece = length <= PAGE_SIZE; @@ -982,13 +980,10 @@ static bool ceph_msg_data_pagelist_advance(struct ceph_msg_data_cursor *cursor, * be processed in that piece. It also tracks whether the current * piece is the last one in the data item. */ -static void ceph_msg_data_cursor_init(struct ceph_msg *msg, size_t length) +static void __ceph_msg_data_cursor_init(struct ceph_msg_data_cursor *cursor) { - struct ceph_msg_data_cursor *cursor = &msg->cursor; - struct ceph_msg_data *data; + size_t length = cursor->total_resid; - data = list_first_entry(&msg->data, struct ceph_msg_data, links); - cursor->data = data; switch (cursor->data->type) { case CEPH_MSG_DATA_PAGELIST: ceph_msg_data_pagelist_cursor_init(cursor, length); @@ -1009,6 +1004,25 @@ static void ceph_msg_data_cursor_init(struct ceph_msg *msg, size_t length) cursor->need_crc = true; } +static void ceph_msg_data_cursor_init(struct ceph_msg *msg, size_t length) +{ + struct ceph_msg_data_cursor *cursor = &msg->cursor; + struct ceph_msg_data *data; + + BUG_ON(!length); + BUG_ON(length > msg->data_length); + BUG_ON(list_empty(&msg->data)); + + data = list_first_entry(&msg->data, struct ceph_msg_data, links); + + cursor->data_head = &msg->data; + cursor->total_resid = length; + data = list_first_entry(&msg->data, struct ceph_msg_data, links); + cursor->data = data; + + __ceph_msg_data_cursor_init(cursor); +} + /* * Return the page containing the next piece to process for a given * data item, and supply the page offset and length of that piece. @@ -1073,8 +1087,16 @@ static bool ceph_msg_data_advance(struct ceph_msg_data_cursor *cursor, BUG(); break; } + cursor->total_resid -= bytes; cursor->need_crc = new_piece; + if (!cursor->resid && cursor->total_resid) { + WARN_ON(!cursor->last_piece); + BUG_ON(list_is_last(&cursor->data->links, cursor->data_head)); + cursor->data = list_entry_next(cursor->data, links); + __ceph_msg_data_cursor_init(cursor); + } + return new_piece; } @@ -2990,8 +3012,6 @@ void ceph_msg_data_set_pages(struct ceph_msg *msg, struct page **pages, BUG_ON(!pages); BUG_ON(!length); - BUG_ON(msg->data_length); - BUG_ON(!list_empty(&msg->data)); data = ceph_msg_data_create(CEPH_MSG_DATA_PAGES); BUG_ON(!data); @@ -3012,8 +3032,6 @@ void ceph_msg_data_set_pagelist(struct ceph_msg *msg, BUG_ON(!pagelist); BUG_ON(!pagelist->length); - BUG_ON(msg->data_length); - BUG_ON(!list_empty(&msg->data)); data = ceph_msg_data_create(CEPH_MSG_DATA_PAGELIST); BUG_ON(!data); @@ -3031,8 +3049,6 @@ void ceph_msg_data_set_bio(struct ceph_msg *msg, struct bio *bio, struct ceph_msg_data *data; BUG_ON(!bio); - BUG_ON(msg->data_length); - BUG_ON(!list_empty(&msg->data)); data = ceph_msg_data_create(CEPH_MSG_DATA_BIO); BUG_ON(!data); diff --git a/include/linux/ceph/messenger.h b/include/linux/ceph/messenger.h index 318da01..de1d2e1 100644 --- a/include/linux/ceph/messenger.h +++ b/include/linux/ceph/messenger.h @@ -108,7 +108,10 @@ struct ceph_msg_data { }; struct ceph_msg_data_cursor { - struct ceph_msg_data *data; /* data item this describes */ + size_t total_resid; /* across all data items */ + struct list_head *data_head; /* = &ceph_msg->data */ + + struct ceph_msg_data *data; /* current data item */ size_t resid; /* bytes not yet consumed */ bool last_piece; /* current is last piece */ bool need_crc; /* crc update needed */ diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c index 8bfe7d3..84703e5 100644 --- a/net/ceph/messenger.c +++ b/net/ceph/messenger.c @@ -734,7 +734,7 @@ static void ceph_msg_data_bio_cursor_init(struct ceph_msg_data_cursor *cursor, BUG_ON(!bio); BUG_ON(!bio->bi_vcnt); - cursor->resid = length; + cursor->resid = min(length, data->bio_length); cursor->bio = bio; cursor->vector_index = 0;