Source: https://patents.google.com/patent/US20070171919A1/en
Timestamp: 2019-06-16 01:10:42
Document Index: 332012049

Matched Legal Cases: ['§ 119', 'Application No. 60', 'Application No. 60', 'Application No. 60', 'Application No. 60', 'Application No. 60']

US20070171919A1 - Message batching with checkpoints systems and methods - Google Patents
Message batching with checkpoints systems and methods Download PDF
US20070171919A1
US20070171919A1 US11/262,314 US26231405A US2007171919A1 US 20070171919 A1 US20070171919 A1 US 20070171919A1 US 26231405 A US26231405 A US 26231405A US 2007171919 A1 US2007171919 A1 US 2007171919A1
US11/262,314
US8238350B2 (en
2004-10-29 Priority to US62384804P priority Critical
2004-11-15 Priority to US62852804P priority
2005-10-28 Priority to US11/262,314 priority patent/US8238350B2/en
2005-10-28 Application filed by Isilon Systems LLC filed Critical Isilon Systems LLC
2006-07-24 Assigned to ISILON SYSTEMS, INC. reassignment ISILON SYSTEMS, INC. ASSIGNMENT OF ASSIGNORS INTEREST (SEE DOCUMENT FOR DETAILS). Assignors: FACHAN, NEAL T., GODMAN, PETER J., PASSEY, AARON J.
2007-07-26 Publication of US20070171919A1 publication Critical patent/US20070171919A1/en
2012-08-07 Publication of US8238350B2 publication Critical patent/US8238350B2/en
The present application claims priority benefit under 35 U.S.C. § 119(e) from U.S. Provisional Application No. 60/623,848, filed Oct. 29, 2004 entitled “Message Batching with Checkpoints Systems and Methods,” and U.S. Provisional Application No. 60/628,528, filed Nov. 15, 2004 entitled “Message Batching with Checkpoints Systems and Methods.” The present application also hereby incorporates by reference herein both of the foregoing applications in their entirety.
The present application relates to U.S. application Ser. No. 11/______ [Attorney Docket ISIL.003A], titled “Non-Blocking Commit Protocol Systems and Methods,” filed on even date herewith, which claims priority to U.S. Provisional Application No. 60/623,843, filed Oct. 29, 2004 entitled “Non-Blocking Commit Protocol Systems and Method;” and U.S. application Ser. No. 11/______ [Attorney Docket ISIL.005A], titled “Distribution System with Asynchronous Execution Systems and Methods,” filed on even date herewith, which claims priority to U.S. Provisional Application No. 60/623,846, filed Oct. 29, 2004 entitled “Distribution System with Asynchronous Execution Systems and Methods,” and U.S. Provisional Application No. 60/628,527, filed Nov. 15, 2004 entitled “Distribution System with Asynchronous Execution Systems and Methods.” The present application hereby incorporates by reference herein all of the foregoing applications in their entirety.
struct rbm_batch_slot_id_t {
efs_devid_t devid;
efs_seqno_t seqno;
int rbm_batch_FOO(rbm_batch_id_t bid, efs_devid_t devid,
struct rbm_batch_slot_id_t *slot_id, u_int opts,
in_args..);
wherein the bid is the returned identifier from the rbm_batch_start function, devid is a device identifier of the destination (i.e., participant node) where the message is to be sent, and in_args define the arguments for the FOO message being sent. As discussed below, the slot_id parameter points to a preallocated memory area where a unique identifier is stored that is later used to lookup data returned by the message. Since update-RPC messages do not return data that needs to be fetched later, they do not have slot_id parameters.
/* Turn off marking the message “critical” (default: on) */
#define RBMB_NONCRITICAL 0x0001
#define RBMB_ALWAYS_RUN 0x0002
#define RBMB_ALWAYS_SEND 0x0004.
Since update_RPC messages do not have responses to communicate a success or failure back to the initiator, they cannot be marked as critical and the opts parameter is omitted. However, in one embodiment, an update_RPC message can be made critical by changing it to an RPC message and discarding the response portion of the RPC message on the participant. This saves a round trip for messages where the normal response is not needed to proceed but an indication of success or failure is desired.
int rbm_batch_lookup_FOO(rbm_batch_id_t bid,
struct rbm_batch_slot_id_t *slot_id,
int *response_code, out_args..);
wherein this lookup function fills out_args with the message response slot identified by the <bid, slot_id> token and frees the batch's internal storage associated with the slot_id. Until the lookup function is called, the response messages are held in the batch as a chain of mbufs in the order they arrived over the network.
This Appendix forms a part of the patent application entitled “MESSAGE BATCHING WITH CHECKPOINTS SYSTEMS AND METHODS,” attorney docket number ISIL.004A.
Generic (initiator and participant) pseudo-code:
efsidp_dispatch
if msg.hdr.batch_id != 0
if msg is response
/* initiator side */
rbm_lbatch_addresp(dev, mbuf)
/* participant side */
efsidp_dispatch_rbatch_msg(dev, mbuf)
[existing dispatch code path]
disconnect_cb(ctx, dev)
/* Take care of initiator side - batches to dev */
foreach batch in g_lbatch
if not (batch.devs contains dev)
foreach batch_msg_ctx in batch.dev.msg_ht
if not ctx.mbuf
ctx.discarded = 1
wakeup ctx
batch.dev.need_collect = 0
batch.dev.wait_count = 0
wakeup(batch.dev.wait_count);
/* Take care of participant side - batches from dev */
foreach batch in g_rbatch
if dev is not initiator
rbm_rbatch_end(dev, bid)
initiator (aka “local”) pseudo-code:
rbm_batch_start(bidp)
batch = alloc new lbatch
batch.id = get next g_lbatch id (never zero)
add_to_hash(g_lbatch.batch_ht, batch.id, batch)
rbm_batch_FOO(bid, devid, slot_id, opts, in_args..)
batch_ctx.flags = 0
batch_ctx.bid = bid
rbmd_usable(devid)
seqno = rbmd_next_seq(devid)
/* CRITICAL is on by default */
if (!(opts & RBMB_NONCRITICAL))
batch_ctx.flags |= IDP_BATCH_CRITICAL
/* slot_id NULL if this is an “update-rpc” message */
rbm_lbatch_addmsg(bid, dev, slot_id, seqno, type,
&batch_ctx.flags)
efsidp_foo(devid, seqno, batch_ctx, in_args..)
rbm_lbatch_cancelmsg(bid, dev, slot_id, seqno)
rbm_lbatch_addmsg(bid, dev, slot_id, seqno, type, flags)
batch = rbm_lbatch_lookup(bid)
if dev not in batch
alloc and add batch_dev to batch
flags |= batch.dev.next_msg_flags
batch.dev.next_msg_flags = 0
if batch.critical_error && !(flags & CLEAR_ERROR)
return ECANCELED
if (slot_id)
/* getting a response */
zero slot_id
/* Add msg to batch's msg_ht */
batch_msg_ctx = alloc;
add_to_hash(batch.dev.msg_ht, seqno, batch_msg_ctx)
slot_id.dev = dev
slot_id.seqno = seqno
/* having a slot implies SEND_RESPONSE */
flags |= IDP_BATCH_SEND_RESPONSE
/* inc count of responses to wait for in collect */
batch.dev.wait_count++
/* we must send a collect message */
batch.dev.need_collect = 1
batch_msg_ctx = del_from_hash(batch.dev.msg_ht, seqno)
free batch_msg_ctx
rbm_lbatch_dec_wait_count(batch, dev)
seqno = hdr.seqno
bid = hdr.batch_id
flags = hdr.batch_flags
batch_msg_ctx = find_in_hash(batch, seqno)
free mbuf
if flags & CRITCAL_ERROR
batch.critical_error = 1
/* TODO flags & CANCEL? how to handle new any-msg-resp? */
batch_msg_ctx.mbuf = mbuf
/* unblock any lookup_FOO for this message */
wakeup(batch_msg_ctx)
rbm_lbatch_dec_wait_count(batch, batch_dev)
batch_dev.wait_count−−
if 0 eq batch_wait_count
/* unblock any collects */
wakeup batch_dev.wait_count
rbm_batch_checkpoint(bid)
foreach dev in batch
batch.dev.next_msg_flags |= CHECKPOINT
/* Only generated for “rpc” messages */
rbm_batch_lookup_FOO(bid, slot_id, response_code, out_args..)
/* may block */
rbm_lbatch_lookup_resp(bid, slot_id, &mbuf)
mbuf_istream_init(mis, mbuf, hdr.size)
rbm_response_code = FOO_resp_unmarshal(mis,
&response_code, ..)
mbuf_istream_clean(mis)
rbm_lbatch_lookup_resp(bid, slot_id, mp)
*mp = 0
dev = slot_id.dev
if dev == 0
return EINVAL
batch_msg_ctx = get_from_hash(batch.dev.msg_ht,
slot_id.seqno)
error = ENOENT
goto out_unlock
while !batch_msg_ctx.mbuf && !batch_msg_ctx.discarded
sleep(batch_msg_ctx, batch->lock)
del_from_hash(batch.dev.msg_ht, slot_id.seqno)
if batch_msg_ctx.discarded
error = ECANCELED
/* pass raw response data back to caller */
mp = &batch_msg_ctx.mbuf
rbm_batch_collect(bid)
rbm_batch_collect_flags(bid, 0)
rbm_batch_collect_flags(bid, flags)
/* send collect messages or wait for msg responses */
/* will be no bigger than this */
slots = alloc slots * (n devs in batch)
n_coll = 0;
/* always clear error on collect */
batch.dev.next_msg_flags |= CLEAR_ERROR
* CLEAR_ERROR implies checkpoint - we want to be
* sure everything before the next msg has completed.
if batch.dev.need_collect
rbm_batch_batch_collect(bid, dev,
&slots[n_coll], 0);
n_coll++
else if batch.dev.wait_count
/* all msgs have responses, wait for them */
while batch.dev.wait_count
sleep(batch.dev.wait_count, batch.lock)
/* wait for any collect message responses */
ASSERT(i < n_coll)
error = lookup_batch_collect(bid,
&slots[i], &code)
if (error || code == ECRITFAIL)
if batch.critical_error
error = ECRITFAIL
batch.critical_error = 0
rbm_batch_end(bid)
/* force sending a collect message */
rbm_batch_collect_flags(bid, FINAL)
/* continue even if rbm_batch_collect fails */
remove from g_lbatch
Participant (aka “remote”) pseudo-code:
if not batch = rbm_rbatch_lookup(dev, bid)
batch = rbm_rbatch_start(dev, bid)
if flags & CHECKPOINT
rbm_rbatch_inc_cp(batch)
/* clear flag - don't do it again if !canrun */
hdr.batch_flags &= {circumflex over ( )}CHECKPOINT
batch.cp_last.msg_cnt++
if rbm_rbatch_canrun(batch, batch.cp_last)
if flags & CLEAR_ERROR
if flags & SEND_RESPONSE
TODO new any-resp fail msg
rbm_rbatch_msg_done(dev, bid)
/* construct batch_ctx for response */
batch_ctx.batch_id = bid
batch_ctx.flags = flags;
if g_msg_info[hdr->type].own_thread
efsidp_dispatch_thread(dev, mbuf,
batch_ctx)
result = handle_FOO(dev, mbuf, batch_ctx)
append mbuf to batch.cp_last.pending
rbm_rbatch_start(dev, bid)
batch = alloc new rbatch(M_ZERO)
batch.id = bid
batch.dev = dev
batch_uid = pack(dev, bid)
add_to_hash(g_rbatch.batch_ht, batch_uid, batch)
/* alloc cp 0 */
rbm_rbatch_lookup(dev, bid)
get_from_hash(g_rbatch.batch_ht, batch_uid)
alloc new cp
cp.id = batch.next_cp_id
batch.next_cp_id++
add_to_hash(batch.cp_ht, cp)
batch.cp_last = cp
rbm_rbatch_advance_cp(batch)
while (batch.cp_active.msg_cnt == 0) and
(batch.cp_active != batch.cp_last)
wakeup batch.cp_active
batch.cp_active = get_from_hash(batch.cp_ht,
batch.cp_active.id + 1)
ASSERT(batch.cp_active)
rbm_rbatch_canrun(batch, cp)
if batch.cp_active.id eq cp.id
/* implies batch.cp_active.id < cp.id */
rbm_rbatch_msg_done(dev, batch_ctx, result)
batch = rbm_rbatch_lookup(dev, ctx.bid)
if result and batch_ctx.flags & CRITICAL
rbm_rbatch_dec_cp_msg(dev, batch)
if ctx.flags & FINAL
rbm_rbatch_dec_cp_msg(batch)
ASSERT(batch.cp_active.msg_cnt)
batch.cp_active.msg_cnt−−
old_cp = batch.cp_active
/* if we advanced the cp, kick off processing */
if old_cp != batch.cp_active
work.action = DWT_BATCH
work.data.bid = batch.id
dwt_dispatch(work)
batch = rbm_rbatch_lookup(bid)
while batch.cp_last.msg_cnt
msleep(batch.cp_last, batch.lock)
remove_from_hash(g_rbatch.dev, batch.id)
dev_local_batch_collect(dev, seq_no, batch_ctx, bid)
send ENOENT
* if we're running, then our collect cp must have been
* reached, so all we have to do is return the status.
* the initiator guarantees no more messages have been
* added to this checkpoint until it receives this collect
response = ECRITFAIL
/* discard batch_ctx to undo munging in efsidp_dispatch */
efsidp_batch_collect_resp(dev, seq_no, NULL, response);
efsidp_FOO_resp(dev, seqno, batch_ctx, args...)
batch_flags = 0
if batch_ctx
batch_id = batch_ctx.batch_id
if response_code and batch_ctx.flags & CRITICAL
batch_flags |= CRITICAL_ERROR
if not batch_ctx or batch_ctx.flags & SEND_RESPONSE
efsid_send(id, type, mos, gi, seqno, batch_ctx)
rbm_rbatch_msg_done(ctx.dev, ctx.batch_ctx, result)
dwt_main
case DWT_BATCH:
rbm_batch_dwt_dispatch(dev, work.bid)
rbm_batch_dwt_dispatch(dev, bid)
batch = rbm_rbatch_lookup(dev, bid)
mbuf = pop(batch.cp_active.mbuf)
/* More to do? Interleave with other dwt work */
if batch.cp_active.mbuf
The following is in the IDP (Isilon Device Protocol) header packet:
struct efsid_header {
rbm_batch_id_t batch_id;
u_int8_t batch_flags;
Where batch_flags may be:
#define IDP_BATCH_CRITICAL 0x01
#define IDP_BATCH_CRITICAL_ERROR 0x02
#define IDP_BATCH_CHECKPOINT 0x04
#define IDP_BATCH_CLEAR_ERROR 0x08
#define IDP_BATCH_SEND_RESPONSE 0x10
#define IDP_BATCH_FINAL 0x20
The following is a pointer parameter to all efsidp_FOO, efsidp_FOO_resp and dev_local_FOO functions:
struct efsidp_batch_ctx {
rbm_batch_id_t batch_id;
u_int8_t batch_flags;
New RBM message for batch collection. Batch ID etc transmitted via efsid_header. Note this is *not* a batchable message, but can be treated like one when processed by the participant.
rpc batch_collect {
callers: async;
handler: async;
options: thread, response-refcount, proto-open;
int response_code;
response-code: response_code;
Generic resp (new type)—
when a CRITICAL message with no SEND_RESPONSE fails
when any type of incoming message with SEND_RESPONSE is discarded on the remote side due to an error occurring on a previous CRITICAL message
This message will take the following format:
update-rpc rpc_cancel {
efs_seqno_t rpc_seqno; # must be set
rbm_batch_id_t batch_id; # zero to ignore
a second storage unit in communication with the first storage unit;
wherein the first storage unit is configured to send a first message and a second message to the second storage unit;
wherein the second message includes a checkpoint flag; and
wherein the second storage unit is configured to selectively execute the first message and the second message in parallel based at least in part on whether the checkpoint flag is set, wherein, if the checkpoint flag is set, the second storage unit executes the first message to completion before executing the second message.
2. The distributed system of claim 1, wherein the first storage unit is further configured to send a third message to the second storage unit requesting a result of executing the first message and the second message.
3. The distributed system of claim 2, wherein the second storage unit waits for the first message and the second message to execute completely before sending the result to the first storage unit.
4. The distributed system of claim 1, wherein at least one of the first message and the second message includes an error flag, and wherein if one or more messages with error flags fail, the first storage unit cancels transmission of subsequent messages to the second storage unit.
5. The distributed system of claim 1, wherein, if one or more messages with error flags fail, the second storage unit discards unexecuted messages.
6. A method for coordinating transactions performed by participating nodes in a network, the method comprising:
sending a plurality of messages from a first node to a second node, wherein at least one message comprises message dependency data;
executing at least a portion of the plurality of messages on the second node, wherein the second node determines when to execute the plurality of messages based at least in part on the message dependency data; and
synchronizing the first node with the second node.
7. The method of claim 6, further comprising generating the dependency data by setting a checkpoint flag in the at least one message.
8. The method of claim 7, wherein determining when to execute the messages on the second node comprises waiting to execute messages received after the checkpoint flag until messages received before receiving the checkpoint flag have been executed to completion.
9. The method of claim 6, further comprising marking at least one of the messages with error handling information.
10. The method of claim 9, wherein if one or more of the at least one marked messages fail, canceling transmission of subsequent messages from the first node to the second node based on the error handling information.
11. The method of claim 9, wherein if one or more of the at least one marked messages fail, discarding unexecuted messages received by the second node based on the error handling information.
12. A method of sending messages between nodes in a communication medium, the method comprising:
transmitting a plurality of messages to one or more nodes, wherein at least one of the messages comprises message ordering information; and
handling the messages in an order based at least in part on the message ordering information.
flagging one or more of the messages; and
handling flagged messages such that if a flagged message fails, no further messages are sent to a node where the flagged message failed, and wherein unexecuted messages already sent to the node where the flagged message failed are discarded.
14. The method of claim 13, further comprising collecting message responses from the one or more nodes after the messages have been processed or discarded.
15. A message initiation system configured to:
transmit a first message to a remote system; and
transmit a second message to the remote system, wherein the second message includes a first flag that indicates that the first message should execute before the second message.
16. The message initiation system of claim 15 wherein at least one of the first message and the second message includes an error flag.
17. The message initiation system of claim 16, wherein the error flag indicates that if the corresponding message fails, messages that have not yet been processed should be discarded.
18. A message participant system configured to:
receive a first message from a message initiation system;
receive a second message from the message initiation system, wherein the second message includes a first flag that indicates that the first message should execute before the second message;
execute the first message; and
after the first message has completed execution, execute the second message.
19. The message participant system of claim 18, wherein at least one of the first message and second message includes an error flag.
20. The message participant system of claim 19, wherein the error flag indicates that if the corresponding message fails, messages that have not yet been processed should be discarded.
US11/262,314 2004-10-29 2005-10-28 Message batching with checkpoints systems and methods Active 2028-12-07 US8238350B2 (en)
US62384804P true 2004-10-29 2004-10-29
US62852804P true 2004-11-15 2004-11-15
US11/262,314 US8238350B2 (en) 2004-10-29 2005-10-28 Message batching with checkpoints systems and methods
US20070171919A1 true US20070171919A1 (en) 2007-07-26
US8238350B2 US8238350B2 (en) 2012-08-07
ID=38285499
US11/262,314 Active 2028-12-07 US8238350B2 (en) 2004-10-29 2005-10-28 Message batching with checkpoints systems and methods
US (1) US8238350B2 (en)
WO2016015039A1 (en) * 2014-07-25 2016-01-28 Netapp, Inc. Asynchronous communications having compounded responses
US9940163B2 (en) * 2015-09-08 2018-04-10 International Business Machines Corporation Ordering repeating elements within a message
US6922696B1 (en) * 2000-05-31 2005-07-26 Sri International Lattice-based security classification system and method
2005-10-28 US US11/262,314 patent/US8238350B2/en active Active
US8953494B2 (en) 2009-09-29 2015-02-10 Juniper Networks, Inc. NETCONF-enabled provisioning in rollback agnostic environment
US9348674B2 (en) 2014-07-25 2016-05-24 Netapp, Inc. Aysnchronous communications having compounded responses
US8238350B2 (en) 2012-08-07
US8700799B2 (en) 2014-04-15 Methods, devices and systems with improved zone merge operation by operating on a switch basis
EP1633109A1 (en) 2006-03-08 Bi-directional process-to-process byte stream protocol
KR101372978B1 (en) 2014-03-13 Managing message queues
EP0814590A2 (en) 1997-12-29 Preventing conflicts in distributed systems
Free format text: ASSIGNMENT OF ASSIGNORS INTEREST;ASSIGNORS:GODMAN, PETER J.;PASSEY, AARON J.;FACHAN, NEAL T.;REEL/FRAME:018114/0765