Skip to content

Commit 29185f5

Browse files
mkeetermkeeter
committed
Minor cleaning
1 parent d064a4f commit 29185f5

File tree

1 file changed

+84
-130
lines changed

1 file changed

+84
-130
lines changed

‎downstairs/src/lib.rs

+84-130
Original file line numberDiff line numberDiff line change
@@ -1182,18 +1182,27 @@ where
11821182
let resp_channel_tx = resp_channel_tx.clone();
11831183
tokio::spawn(async move {
11841184
while let Some(m) = message_channel_rx.recv().await {
1185-
if let Err(e) = Downstairs::proc_frame(
1185+
match Downstairs::proc_frame(
11861186
&adc,
11871187
upstairs_connection,
11881188
m,
11891189
&resp_channel_tx,
1190-
&tx,
11911190
)
11921191
.await
11931192
{
1194-
bail!("Proc frame returns error: {}", e);
1193+
// If we added work, tell the work task to get busy.
1194+
Ok(Some(new_ds_id)) => {
1195+
cdt::work__start!(|| new_ds_id.0);
1196+
tx.send(()).await?;
1197+
}
1198+
// If we handled the job locally, nothing to do here
1199+
Ok(None) => (),
1200+
Err(e) => {
1201+
bail!("Proc frame returns error: {}", e);
1202+
}
11951203
}
11961204
}
1205+
11971206
Ok(())
11981207
})
11991208
};
@@ -2231,26 +2240,62 @@ impl Downstairs {
22312240
Ok(())
22322241
}
22332242

2234-
/*
2235-
* A new IO request has been received.
2236-
* If the message is a ping, send the correct response. If the message is an
2237-
* IO, then put the new IO the work hashmap. If the message is a repair
2238-
* message, then we handle it right here.
2239-
*/
2243+
/// Handle a new message from the upstairs
2244+
///
2245+
/// If the message is an IO, then put the new IO the work hashmap. If the
2246+
/// message is a repair message, then we handle it right here.
22402247
async fn proc_frame(
22412248
ad: &Mutex<Downstairs>,
22422249
upstairs_connection: UpstairsConnection,
22432250
m: Message,
22442251
resp_tx: &mpsc::Sender<Message>,
2245-
job_channel_tx: &mpsc::Sender<()>,
2246-
) -> Result<()> {
2247-
let new_ds_id = match m {
2252+
) -> Result<Option<JobId>> {
2253+
// Initial check against upstairs and session ID
2254+
match m {
22482255
Message::Write {
22492256
upstairs_id,
22502257
session_id,
2251-
job_id,
2252-
dependencies,
2253-
writes,
2258+
..
2259+
}
2260+
| Message::WriteUnwritten {
2261+
upstairs_id,
2262+
session_id,
2263+
..
2264+
}
2265+
| Message::Flush {
2266+
upstairs_id,
2267+
session_id,
2268+
..
2269+
}
2270+
| Message::ReadRequest {
2271+
upstairs_id,
2272+
session_id,
2273+
..
2274+
}
2275+
| Message::ExtentLiveClose {
2276+
upstairs_id,
2277+
session_id,
2278+
..
2279+
}
2280+
| Message::ExtentLiveFlushClose {
2281+
upstairs_id,
2282+
session_id,
2283+
..
2284+
}
2285+
| Message::ExtentLiveRepair {
2286+
upstairs_id,
2287+
session_id,
2288+
..
2289+
}
2290+
| Message::ExtentLiveReopen {
2291+
upstairs_id,
2292+
session_id,
2293+
..
2294+
}
2295+
| Message::ExtentLiveNoOp {
2296+
upstairs_id,
2297+
session_id,
2298+
..
22542299
} => {
22552300
if !is_message_valid(
22562301
upstairs_connection,
@@ -2260,8 +2305,19 @@ impl Downstairs {
22602305
)
22612306
.await?
22622307
{
2263-
return Ok(());
2308+
return Ok(None);
22642309
}
2310+
}
2311+
_ => (),
2312+
}
2313+
2314+
let r = match m {
2315+
Message::Write {
2316+
job_id,
2317+
dependencies,
2318+
writes,
2319+
..
2320+
} => {
22652321
cdt::submit__write__start!(|| job_id.0);
22662322

22672323
let new_write = IOop::Write {
@@ -2274,25 +2330,14 @@ impl Downstairs {
22742330
Some(job_id)
22752331
}
22762332
Message::Flush {
2277-
upstairs_id,
2278-
session_id,
22792333
job_id,
22802334
dependencies,
22812335
flush_number,
22822336
gen_number,
22832337
snapshot_details,
22842338
extent_limit,
2339+
..
22852340
} => {
2286-
if !is_message_valid(
2287-
upstairs_connection,
2288-
upstairs_id,
2289-
session_id,
2290-
resp_tx,
2291-
)
2292-
.await?
2293-
{
2294-
return Ok(());
2295-
}
22962341
cdt::submit__flush__start!(|| job_id.0);
22972342

22982343
let new_flush = IOop::Flush {
@@ -2308,22 +2353,11 @@ impl Downstairs {
23082353
Some(job_id)
23092354
}
23102355
Message::WriteUnwritten {
2311-
upstairs_id,
2312-
session_id,
23132356
job_id,
23142357
dependencies,
23152358
writes,
2359+
..
23162360
} => {
2317-
if !is_message_valid(
2318-
upstairs_connection,
2319-
upstairs_id,
2320-
session_id,
2321-
resp_tx,
2322-
)
2323-
.await?
2324-
{
2325-
return Ok(());
2326-
}
23272361
cdt::submit__writeunwritten__start!(|| job_id.0);
23282362

23292363
let new_write = IOop::WriteUnwritten {
@@ -2336,22 +2370,11 @@ impl Downstairs {
23362370
Some(job_id)
23372371
}
23382372
Message::ReadRequest {
2339-
upstairs_id,
2340-
session_id,
23412373
job_id,
23422374
dependencies,
23432375
requests,
2376+
..
23442377
} => {
2345-
if !is_message_valid(
2346-
upstairs_connection,
2347-
upstairs_id,
2348-
session_id,
2349-
resp_tx,
2350-
)
2351-
.await?
2352-
{
2353-
return Ok(());
2354-
}
23552378
cdt::submit__read__start!(|| job_id.0);
23562379

23572380
let new_read = IOop::Read {
@@ -2365,23 +2388,11 @@ impl Downstairs {
23652388
}
23662389
// These are for repair while taking live IO
23672390
Message::ExtentLiveClose {
2368-
upstairs_id,
2369-
session_id,
23702391
job_id,
23712392
dependencies,
23722393
extent_id,
2394+
..
23732395
} => {
2374-
if !is_message_valid(
2375-
upstairs_connection,
2376-
upstairs_id,
2377-
session_id,
2378-
resp_tx,
2379-
)
2380-
.await?
2381-
{
2382-
return Ok(());
2383-
}
2384-
23852396
cdt::submit__el__close__start!(|| job_id.0);
23862397
// TODO: Add dtrace probes
23872398
let ext_close = IOop::ExtentClose {
@@ -2394,25 +2405,13 @@ impl Downstairs {
23942405
Some(job_id)
23952406
}
23962407
Message::ExtentLiveFlushClose {
2397-
upstairs_id,
2398-
session_id,
23992408
job_id,
24002409
dependencies,
24012410
extent_id,
24022411
flush_number,
24032412
gen_number,
2413+
..
24042414
} => {
2405-
if !is_message_valid(
2406-
upstairs_connection,
2407-
upstairs_id,
2408-
session_id,
2409-
resp_tx,
2410-
)
2411-
.await?
2412-
{
2413-
return Ok(());
2414-
}
2415-
24162415
cdt::submit__el__flush__close__start!(|| job_id.0);
24172416
// Do both the flush, and then the close
24182417
let new_flush = IOop::ExtentFlushClose {
@@ -2427,25 +2426,12 @@ impl Downstairs {
24272426
Some(job_id)
24282427
}
24292428
Message::ExtentLiveRepair {
2430-
upstairs_id,
2431-
session_id,
24322429
job_id,
24332430
dependencies,
24342431
extent_id,
24352432
source_repair_address,
24362433
..
24372434
} => {
2438-
if !is_message_valid(
2439-
upstairs_connection,
2440-
upstairs_id,
2441-
session_id,
2442-
resp_tx,
2443-
)
2444-
.await?
2445-
{
2446-
return Ok(());
2447-
}
2448-
24492435
cdt::submit__el__repair__start!(|| job_id.0);
24502436
// Do both the flush, and then the close
24512437
let new_repair = IOop::ExtentLiveRepair {
@@ -2461,23 +2447,11 @@ impl Downstairs {
24612447
Some(job_id)
24622448
}
24632449
Message::ExtentLiveReopen {
2464-
upstairs_id,
2465-
session_id,
24662450
job_id,
24672451
dependencies,
24682452
extent_id,
2453+
..
24692454
} => {
2470-
if !is_message_valid(
2471-
upstairs_connection,
2472-
upstairs_id,
2473-
session_id,
2474-
resp_tx,
2475-
)
2476-
.await?
2477-
{
2478-
return Ok(());
2479-
}
2480-
24812455
cdt::submit__el__reopen__start!(|| job_id.0);
24822456
let new_open = IOop::ExtentLiveReopen {
24832457
dependencies,
@@ -2489,21 +2463,10 @@ impl Downstairs {
24892463
Some(job_id)
24902464
}
24912465
Message::ExtentLiveNoOp {
2492-
upstairs_id,
2493-
session_id,
24942466
job_id,
24952467
dependencies,
2468+
..
24962469
} => {
2497-
if !is_message_valid(
2498-
upstairs_connection,
2499-
upstairs_id,
2500-
session_id,
2501-
resp_tx,
2502-
)
2503-
.await?
2504-
{
2505-
return Ok(());
2506-
}
25072470
cdt::submit__el__noop__start!(|| job_id.0);
25082471
let new_open = IOop::ExtentLiveNoOp { dependencies };
25092472

@@ -2551,7 +2514,7 @@ impl Downstairs {
25512514
}
25522515
};
25532516
resp_tx.send(msg).await?;
2554-
return Ok(());
2517+
None
25552518
}
25562519
Message::ExtentClose {
25572520
repair_id,
@@ -2570,7 +2533,7 @@ impl Downstairs {
25702533
}
25712534
};
25722535
resp_tx.send(msg).await?;
2573-
return Ok(());
2536+
None
25742537
}
25752538
Message::ExtentRepair {
25762539
repair_id,
@@ -2604,7 +2567,7 @@ impl Downstairs {
26042567
}
26052568
};
26062569
resp_tx.send(msg).await?;
2607-
return Ok(());
2570+
None
26082571
}
26092572
Message::ExtentReopen {
26102573
repair_id,
@@ -2623,20 +2586,11 @@ impl Downstairs {
26232586
}
26242587
};
26252588
resp_tx.send(msg).await?;
2626-
return Ok(());
2589+
None
26272590
}
26282591
x => bail!("unexpected frame {:?}", x),
26292592
};
2630-
2631-
/*
2632-
* If we added work, tell the work task to get busy.
2633-
*/
2634-
if let Some(new_ds_id) = new_ds_id {
2635-
cdt::work__start!(|| new_ds_id.0);
2636-
job_channel_tx.send(()).await?;
2637-
}
2638-
2639-
Ok(())
2593+
Ok(r)
26402594
}
26412595

26422596
async fn do_work_for(

0 commit comments

Comments
 (0)