[RSDK-8798] - Async save command #15

seanavery · 2024-09-23T18:00:52Z

Description

This PR adds an async option to the save do_command to allow users to fire of save commands that include the most recent video segment. Right now this simply waits the segment duration - this could be further optimized by adding a fs watcher to execute the save immediately when the segment is finished.

Executes async save command in asynchronous go routine that waits for segment duration before executing.
Refactors filename handling for concat calls.
- Separates out concat mp4 file name generation into a helper.
- Move to writing concat txt files to /tmp path.
- Generates unique concat txt files with a uuid to allow for concurrent operations.

Test

Async save cmd ✅
Sync save cmd ✅
Fetch cmd still works ✅
Cleanup helper removes stale conact txt files if leftover ✅
Added unit test for async save cmd ✅
Added unit test for cleaning up leftover concat txt files ✅

README.md

randhid · 2024-09-24T10:25:56Z

cam/cam.go

+		uploadFilePath := generateOutputFilePath(vs.name.Name, formatDateTimeToString(from), metadata, vs.uploadPath)
+		uploadFileName := filepath.Base(uploadFilePath)


with a bit of a refactor to have vs embed a resource.Named interface and passing it conf.Name.AsNamed() in the constructor/reconfiguration, you could shorten getting the name to a vs.Name call. Also, test these two functions to make sure they're giving you what you expect.

Yeah good call out. Looks like conf.ResourceName().AsNamed() returns Named, but I really just care about the string value.

Shouldn't I just pull out the string with conf.ResourceName().Name directly into the struct?

To satisfy the Camera interface, it looks like I need a method on vs Name that returns resource.Name

You can pull it out with .ShortName I think, but if it's not following what I think other resource interfaces follow, then we can just leave it be.

cam/concater.go

randhid · 2024-09-24T10:35:34Z

cam/utils.go

+func validateSaveCommand(command map[string]interface{}) (time.Time, time.Time, string, bool, error) {
 	fromStr, ok := command["from"].(string)
 	if !ok {
-		return time.Time{}, time.Time{}, "", errors.New("from timestamp not found")
+		return time.Time{}, time.Time{}, "", false, errors.New("from timestamp not found")
 	}
 	from, err := parseDateTimeString(fromStr)
 	if err != nil {
-		return time.Time{}, time.Time{}, "", err
+		return time.Time{}, time.Time{}, "", false, err
 	}
 	toStr, ok := command["to"].(string)
 	if !ok {
-		return time.Time{}, time.Time{}, "", errors.New("to timestamp not found")
+		return time.Time{}, time.Time{}, "", false, errors.New("to timestamp not found")
 	}
 	to, err := parseDateTimeString(toStr)
 	if err != nil {
-		return time.Time{}, time.Time{}, "", err
+		return time.Time{}, time.Time{}, "", false, err
 	}
 	if from.After(to) {
-		return time.Time{}, time.Time{}, "", errors.New("from timestamp is after to timestamp")
+		return time.Time{}, time.Time{}, "", false, errors.New("from timestamp is after to timestamp")
 	}
 	metadata, ok := command["metadata"].(string)
 	if !ok {
 		metadata = ""
 	}
-	return from, to, metadata, nil
+	async, ok := command["async"].(bool)
+	if !ok {
+		async = false
+	}
+	return from, to, metadata, async, nil


teh returns here are getting pretty long, that's okay, but you may want to break this up into separate validation functions if you need more in the future or return a struct or use a function in the scope of DoCommand. Nothing to do now.

tests/save_test.go

cam/cam.go

tests/save_test.go

hexbabe

I'm slightly afraid that long segment durations could cause gRPC timeouts. Have we tested long segment durations with the various SDKs?

cam/cam.go

hexbabe · 2024-09-24T14:06:34Z

tests/save_test.go

@@ -112,6 +112,15 @@ func TestSaveDoCommand(t *testing.T) {
 		"metadata": "test-metadata",
 	}

+	// Valid async save


Super optional: but looking ahead, we could probably add a test now that tests a time range that overlaps with the current time/the segment currently being written to.

Prob an important test to test the future feature of not waiting the entire segment duration every time.

Yes I should add such a test. Running into issue where viam-server crashes when waiting for the async call in the test run - will try to get it working tmrw.

Ah, I cannot seem to get around an issue where my module crashes when sleeping in a test case. Not sure at this point why waiting in a test is interfering with the robot runtime.

\_ Assertion ((src_linesize) >= 0 ? (src_linesize) : (-(src_linesize))) >= bytewidth failed at libavutil/imgutils.c:350 2024-09-25T16:27:04.727Z ERROR video-store-module.modmanager.process.video-storage_/host/bin/video-store.StdErr pexec/managed_process.go:275 \_ SIGABRT: abort

I have tried a few strategies including throwing sleep in a routine with async wait group with no success.

Would you be ok if I throw this into the test improvements ticket?

Yeah that sounds good to me. Not blocking

randhid · 2024-09-24T21:59:37Z

cam/cam.go

+			vs.workers.Add(func(ctx context.Context) {
+				vs.asyncSave(ctx, from, to, uploadFilePath)


Things to consider:
1 - We add a worker with each DoCommand request, do we want them to interrupt already working workers?
2 - Should another call interrupt this worker?
3 - workers.Stop will stop all workers so if you want them to be separate from your other workers, give this new set of workers a new variable in the struct.

1 - We add a worker with each DoCommand request, do we want them to interrupt already working workers?

I do not want the do_command workers to interfere with the other long-running workers (processFrames, deleter). Adding workers should not interfere:

type StoppableWorkers struct { // Use a `sync.RWMutex` isntead of a `sync.Mutex` so that additions of new // workers do not lock with each other in any way. We want // as-fast-as-possible worker addition. mu sync.RWMutex ctx context.Context cancelFunc func() workers sync.WaitGroup }

2 - Should another call interrupt this worker?

The short-lived asynSave routine should just die out when finished through the cancel call.

workers.Stop will stop all workers so if you want them to be separate from your other workers, give this new set of workers a new variable in the struct.

It is ok to stop all workers during close - processFrames, deleter, and all outstanding async calls can be shutdown together.

seanavery added 8 commits September 20, 2024 16:52

Hacky async save

ef8320b

Return upload filename in async resp

7d25255

Remove file name setter from concat

a02271b

Cleanup file setters

0f96a9d

Add comments

3e6e86a

Add cleanup for leftover concat txt files

e5d49ef

Add async save unit test

fb50e68

Spell check

03ea3cc

seanavery requested review from randhid and hexbabe and removed request for randhid September 23, 2024 19:20

seanavery added 2 commits September 23, 2024 16:27

Add async save to readme

81d3aa4

Format

76e774a

randhid requested changes Sep 24, 2024

View reviewed changes

hexbabe reviewed Sep 24, 2024

View reviewed changes

cam/cam.go Outdated Show resolved Hide resolved

cam/cam.go Outdated Show resolved Hide resolved

hexbabe reviewed Sep 24, 2024

View reviewed changes

seanavery added 4 commits September 24, 2024 16:11

Add test for concat txt cleanup

5ea33fc

Use switch statement

242a187

Use workers for asyncs save

e5f926e

Add debug logs to async concat

a95f5f2

randhid reviewed Sep 24, 2024

View reviewed changes

seanavery requested review from randhid and hexbabe September 25, 2024 17:23

hexbabe approved these changes Sep 25, 2024

View reviewed changes

randhid approved these changes Sep 25, 2024

View reviewed changes

seanavery merged commit b9d2829 into main Sep 25, 2024
4 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[RSDK-8798] - Async save command #15

[RSDK-8798] - Async save command #15

seanavery commented Sep 23, 2024 •

edited

Loading

randhid Sep 24, 2024

seanavery Sep 24, 2024

seanavery Sep 24, 2024

randhid Sep 24, 2024

randhid Sep 24, 2024

hexbabe left a comment

hexbabe Sep 24, 2024

seanavery Sep 24, 2024 •

edited

Loading

hexbabe Sep 25, 2024

seanavery Sep 25, 2024

hexbabe Sep 25, 2024

randhid Sep 24, 2024

seanavery Sep 25, 2024

		uploadFilePath := generateOutputFilePath(vs.name.Name, formatDateTimeToString(from), metadata, vs.uploadPath)
		uploadFileName := filepath.Base(uploadFilePath)

		vs.workers.Add(func(ctx context.Context) {
		vs.asyncSave(ctx, from, to, uploadFilePath)

[RSDK-8798] - Async save command #15

[RSDK-8798] - Async save command #15

Conversation

seanavery commented Sep 23, 2024 • edited Loading

Description

Test

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hexbabe left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

seanavery Sep 24, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

seanavery commented Sep 23, 2024 •

edited

Loading

seanavery Sep 24, 2024 •

edited

Loading