Skip to content

storage_write: track nbytespending per connection #661

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

storage_write: track nbytespending per connection #661

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3b64974
storage_write.c: store connection number in cookie
gperciva Mar 28, 2025
9f4fac7
storage_write.c: track nbytespending per connection
gperciva Mar 28, 2025
c4ac3be
REBASE? sanity check?
gperciva Mar 28, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 23 additions & 10 deletions tar/storage/storage_write.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ struct storage_write_internal {
size_t lastcnum;

/* Number of bytes of pending writes. */
size_t nbytespending;
size_t nbytespending[AGGRESSIVE_CNUM];

/* Last time we wrote a checkpoint. */
uint64_t lastcheckpoint;
Expand All @@ -66,6 +66,9 @@ struct write_file_internal {
uint64_t machinenum;
int done;

/* Connection we're using. */
size_t conn;

/* Parameters used in write_file. */
uint8_t class;
uint8_t name[32];
Expand Down Expand Up @@ -144,7 +147,8 @@ storage_write_start(uint64_t machinenum, const uint8_t lastseq[32],
S->lastcnum = 0;

/* No pending writes so far. */
S->nbytespending = 0;
for (i = 0; i < S->numconns; i++)
S->nbytespending[i] = 0;

/* No checkpoint yet. */
S->lastcheckpoint = 0;
Expand Down Expand Up @@ -334,21 +338,23 @@ storage_write_file(STORAGE_W * S, uint8_t * buf, size_t len,
if (crypto_file_enc(buf, len, C->filebuf))
goto err2;

/* Select connection to use. */
C->conn = S->lastcnum = (S->lastcnum + 1) % S->numconns;

/* We're issuing a write operation. */
S->nbytespending += C->flen;
S->nbytespending[C->conn] += C->flen;

/*
* Make sure the pending operation queue isn't too large before we
* add yet another operation to it.
*/
while (S->nbytespending > MAXPENDING_WRITEBYTES) {
while (S->nbytespending[C->conn] > MAXPENDING_WRITEBYTES) {
if (network_select(1))
goto err2;
}

/* Ask the netpacket layer to send a request and get a response. */
S->lastcnum = (S->lastcnum + 1) % S->numconns;
if (netpacket_op(S->NPC[S->lastcnum], callback_write_file_send, C))
if (netpacket_op(S->NPC[C->conn], callback_write_file_send, C))
goto err0;

/* Send ourself SIGQUIT or SIGUSR2 if necessary. */
Expand Down Expand Up @@ -414,7 +420,11 @@ callback_write_file_response(void * cookie,
switch (packetbuf[0]) {
case 0:
/* This write operation is no longer pending. */
C->S->nbytespending -= C->flen;
if (C->S->nbytespending[C->conn] < C->flen) {
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is probably a good idea?

I mean, other than random memory corruption, I can't think of why this might occur. Now, if we have random corruption, then this check could be useful... but likely a whole bunch of other things would break anyway. shrug

warn0("Invalid connection number and/or length");
goto err1;
}
C->S->nbytespending[C->conn] -= C->flen;
break;
case 1:
warn0("Cannot store file: File already exists");
Expand Down Expand Up @@ -462,15 +472,18 @@ callback_write_file_response(void * cookie,
int
storage_write_flush(STORAGE_W * S)
{
size_t i;

/* No-op on NULL. */
if (S == NULL)
return (0);

/* Wait until all pending writes have been completed. */
while (S->nbytespending > 0) {
if (network_select(1))
goto err0;
for (i = 0; i < S->numconns; i++) {
while (S->nbytespending[i] > 0) {
if (network_select(1))
goto err0;
}
}

/* Success! */
Expand Down