Skip to content
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

support streaming snapshot over fd #39

Merged
merged 3 commits into from
Apr 28, 2023
Merged

support streaming snapshot over fd #39

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
399126f
feat: handle snapshot write size
mhofman Sep 30, 2022
1959e1c
feat: write snapshot to fd
mhofman Apr 27, 2023
9d408aa
feat: read snapshot from fd
mhofman Apr 27, 2023
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
50 changes: 43 additions & 7 deletions xsnap/sources/xsnap-worker.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,14 +93,22 @@ xsCallback gxSnapshotCallbacks[mxSnapshotCallbackCount] = {
// fx_clearTimer,
};

typedef struct {
FILE *file;
int size;
} SnapshotStream;

static int fxSnapshotRead(void* stream, void* address, size_t size)
{
return (fread(address, size, 1, stream) == 1) ? 0 : errno;
}

static int fxSnapshotWrite(void* stream, void* address, size_t size)
{
return (fwrite(address, size, 1, stream) == 1) ? 0 : errno;
SnapshotStream* snapshotStream = stream;
size_t written = fwrite(address, size, 1, snapshotStream->file);
snapshotStream->size += size * written;
return (written == 1) ? 0 : errno;
}

#if mxInstrument
Expand Down Expand Up @@ -331,7 +339,18 @@ int main(int argc, char* argv[])
}
xsInitializeSharedCluster();
if (argr) {
snapshot.stream = fopen(argv[argr], "rb");
char *path = argv[argr];
if (path[0] == '@') {
int fd = atoi(path + 1);
int tmpfd = dup(fd);
Copy link
Collaborator

Choose a reason for hiding this comment

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

What is the dup() for? Does it allow us to fclose(stream.file) but then re-use the original fd later? Do we have a use for that, or is this a generic pattern that we do need for the writing side?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

We do need it for the writing side, and I figured in the future we may want the ability to reuse the load snapshot stream.

if (tmpfd < 0) {
snapshot.stream = NULL;
} else {
snapshot.stream = fdopen(tmpfd, "rb");
}
} else {
snapshot.stream = fopen(path, "rb");
}
if (snapshot.stream) {
machine = xsReadSnapshot(&snapshot, "xsnap", NULL);
fclose(snapshot.stream);
Expand Down Expand Up @@ -528,10 +547,24 @@ int main(int argc, char* argv[])
fxTestRecord(mxTestRecordParam, nsbuf + 1, nslen - 1);
#endif
path = nsbuf + 1;
snapshot.stream = fopen(path, "wb");
if (snapshot.stream) {
SnapshotStream stream;
if (path[0] == '@') {
int fd = atoi(path + 1);
int tmpfd = dup(fd);
if (tmpfd < 0) {
stream.file = NULL;
} else {
stream.file = fdopen(tmpfd, "ab");
}
} else {
stream.file = fopen(path, "wb");
}
stream.size = 0;
if (stream.file) {
snapshot.stream = &stream;
fxWriteSnapshot(machine, &snapshot);
fclose(snapshot.stream);
snapshot.stream = NULL;
fclose(stream.file);
}
else
snapshot.error = errno;
Expand All @@ -541,7 +574,10 @@ int main(int argc, char* argv[])
c_exit(E_IO_ERROR);
}
if (snapshot.error == 0) {
int writeError = fxWriteOkay(toParent, meterIndex, machine, "", 0);
// Allows us to format up to 999,999,999,999 bytes (1TiB - 1)
char fsize[13];
int fsizeLength = snprintf(fsize, sizeof(fsize), "%d", stream.size);
int writeError = fxWriteOkay(toParent, meterIndex, machine, fsize, fsizeLength);
if (writeError != 0) {
fprintf(stderr, "%s\n", fxWriteNetStringError(writeError));
c_exit(E_IO_ERROR);
Expand Down Expand Up @@ -810,7 +846,7 @@ static int fxWriteOkay(FILE* outStream, xsUnsignedValue meterIndex, xsMachine *t
char numeral64[] = "12345678901234567890"; // big enough for 64bit numeral
char prefix[8 + sizeof fmt + 8 * sizeof numeral64 + sizeof timestampBuffer];
// Prepend the meter usage to the reply.
snprintf(prefix, sizeof(prefix) - 1, fmt,
snprintf(prefix, sizeof(prefix), fmt,
fxGetCurrentHeapCount(the),
meterIndex, the->allocatedSpace, tsbuf);
return fxWriteNetString(outStream, prefix, buf, length);
Expand Down