ooni-probe-cli/pkg/oonimkall/task.go

84 lines
2.3 KiB
Go
Raw Permalink Normal View History

package oonimkall
import (
"context"
"encoding/json"
refactor: flatten and separate (#353) * refactor(atomicx): move outside the engine package After merging probe-engine into probe-cli, my impression is that we have too much unnecessary nesting of packages in this repository. The idea of this commit and of a bunch of following commits will instead be to reduce the nesting and simplify the structure. While there, improve the documentation. * fix: always use the atomicx package For consistency, never use sync/atomic and always use ./internal/atomicx so we can just grep and make sure we're not risking to crash if we make a subtle mistake on a 32 bit platform. While there, mention in the contributing guidelines that we want to always prefer the ./internal/atomicx package over sync/atomic. * fix(atomicx): remove unnecessary constructor We don't need a constructor here. The default constructed `&Int64{}` instance is already usable and the constructor does not add anything to what we are doing, rather it just creates extra confusion. * cleanup(atomicx): we are not using Float64 Because atomicx.Float64 is unused, we can safely zap it. * cleanup(atomicx): simplify impl and improve tests We can simplify the implementation by using defer and by letting the Load() method call Add(0). We can improve tests by making many goroutines updated the atomic int64 value concurrently. * refactor(fsx): can live in the ./internal pkg Let us reduce the amount of nesting. While there, ensure that the package only exports the bare minimum, and improve the documentation of the tests, to ease reading the code. * refactor: move runtimex to ./internal * refactor: move shellx into the ./internal package While there, remove unnecessary dependency between packages. While there, specify in the contributing guidelines that one should use x/sys/execabs instead of os/exec. * refactor: move ooapi into the ./internal pkg * refactor(humanize): move to ./internal and better docs * refactor: move platform to ./internal * refactor(randx): move to ./internal * refactor(multierror): move into the ./internal pkg * refactor(kvstore): all kvstores in ./internal Rather than having part of the kvstore inside ./internal/engine/kvstore and part in ./internal/engine/kvstore.go, let us put every piece of code that is kvstore related into the ./internal/kvstore package. * fix(kvstore): always return ErrNoSuchKey on Get() error It should help to use the kvstore everywhere removing all the copies that are lingering around the tree. * sessionresolver: make KVStore mandatory Simplifies implementation. While there, use the ./internal/kvstore package rather than having our private implementation. * fix(ooapi): use the ./internal/kvstore package * fix(platform): better documentation
2021-06-04 10:34:18 +02:00
"github.com/ooni/probe-cli/v3/internal/atomicx"
"github.com/ooni/probe-cli/v3/internal/runtimex"
)
// Task is an asynchronous task running an experiment. It mimics the
// namesake concept initially implemented in Measurement Kit.
//
// Future directions
//
// Currently Task and Session are two unrelated APIs. As part of
// evolving the APIs with which apps interact with the engine, we
// will modify Task to run in the context of a Session. We will
// do that to save extra lookups and to allow several experiments
// running as subsequent Tasks to reuse the Session connections
// created with the OONI probe services backends.
type Task struct {
fix(oonimkall): improve channel management pattern (#621) 1. add eof channel to event emitter and use this channel as signal that we shouldn't be sending anymore instead of using a pattern where we use a timer to decide sending has timed out (because we're using a buffered channel, it is still possible for some evetns to end up in the channel if there is space, but this is not a concern, because the events will be deleted when the channel itself is gone); 2. refactor all tests where we assumed the output channel was closed to actually use a parallel "eof" channel and use it as signal we should not be sending anymore (not strictly required but still the right thing to do in terms of using consistent patterns); 3. modify how we construct a runner so that it passes to the event emitter an "eof" channel and closes this channel when the main goroutine running the task is terminating; 4. modify the task to signal events such as "task goroutine started" and "task goroutine stopped" using channels, which helps to write much more correct tests; 5. take advantage of the previous change to improve the test that ensures we're not blocking for a small number of events and also improve the name of such a test to reflect what it's testing. The related issue in term of fixing the channel usage pattern is https://github.com/ooni/probe/issues/1438. Regarding improving testability, instead, the correct reference issue is https://github.com/ooni/probe/issues/1903. There are possibly more changes to apply here to improve this package and its testability, but let's land this diff first and then see how much time is left for further improvements. I've run unit and integration tests with `-race` locally. This diff will need to be backported to `release/3.11`.
2021-12-01 15:40:25 +01:00
cancel context.CancelFunc
isdone *atomicx.Int64
isstarted chan interface{} // for testing
isstopped chan interface{} // for testing
out chan *event
}
// StartTask starts an asynchronous task. The input argument is a
// serialized JSON conforming to MK v0.10.9's API.
func StartTask(input string) (*Task, error) {
var settings settings
if err := json.Unmarshal([]byte(input), &settings); err != nil {
return nil, err
}
const bufsiz = 128 // common case: we don't want runner to block
ctx, cancel := context.WithCancel(context.Background())
task := &Task{
fix(oonimkall): improve channel management pattern (#621) 1. add eof channel to event emitter and use this channel as signal that we shouldn't be sending anymore instead of using a pattern where we use a timer to decide sending has timed out (because we're using a buffered channel, it is still possible for some evetns to end up in the channel if there is space, but this is not a concern, because the events will be deleted when the channel itself is gone); 2. refactor all tests where we assumed the output channel was closed to actually use a parallel "eof" channel and use it as signal we should not be sending anymore (not strictly required but still the right thing to do in terms of using consistent patterns); 3. modify how we construct a runner so that it passes to the event emitter an "eof" channel and closes this channel when the main goroutine running the task is terminating; 4. modify the task to signal events such as "task goroutine started" and "task goroutine stopped" using channels, which helps to write much more correct tests; 5. take advantage of the previous change to improve the test that ensures we're not blocking for a small number of events and also improve the name of such a test to reflect what it's testing. The related issue in term of fixing the channel usage pattern is https://github.com/ooni/probe/issues/1438. Regarding improving testability, instead, the correct reference issue is https://github.com/ooni/probe/issues/1903. There are possibly more changes to apply here to improve this package and its testability, but let's land this diff first and then see how much time is left for further improvements. I've run unit and integration tests with `-race` locally. This diff will need to be backported to `release/3.11`.
2021-12-01 15:40:25 +01:00
cancel: cancel,
isdone: &atomicx.Int64{},
isstarted: make(chan interface{}),
isstopped: make(chan interface{}),
out: make(chan *event, bufsiz),
}
go func() {
fix(oonimkall): improve channel management pattern (#621) 1. add eof channel to event emitter and use this channel as signal that we shouldn't be sending anymore instead of using a pattern where we use a timer to decide sending has timed out (because we're using a buffered channel, it is still possible for some evetns to end up in the channel if there is space, but this is not a concern, because the events will be deleted when the channel itself is gone); 2. refactor all tests where we assumed the output channel was closed to actually use a parallel "eof" channel and use it as signal we should not be sending anymore (not strictly required but still the right thing to do in terms of using consistent patterns); 3. modify how we construct a runner so that it passes to the event emitter an "eof" channel and closes this channel when the main goroutine running the task is terminating; 4. modify the task to signal events such as "task goroutine started" and "task goroutine stopped" using channels, which helps to write much more correct tests; 5. take advantage of the previous change to improve the test that ensures we're not blocking for a small number of events and also improve the name of such a test to reflect what it's testing. The related issue in term of fixing the channel usage pattern is https://github.com/ooni/probe/issues/1438. Regarding improving testability, instead, the correct reference issue is https://github.com/ooni/probe/issues/1903. There are possibly more changes to apply here to improve this package and its testability, but let's land this diff first and then see how much time is left for further improvements. I've run unit and integration tests with `-race` locally. This diff will need to be backported to `release/3.11`.
2021-12-01 15:40:25 +01:00
close(task.isstarted)
forwardport: pull the patches mentioned in ooni/probe#1908 (#629) * [forwardport] fix(oonimkall): make logger used by tasks unit testable (#623) This diff forward ports e4b04642c51e7461728b25941624e1b97ef0ec83. Reference issue: https://github.com/ooni/probe/issues/1903 * [forwardport] feat(oonimkall): improve taskEmitter testability (#624) This diff forward ports 3e0f01a389c1f4cdd7878ec151aff91870a0bdff. 1. rename eventemitter{,_test}.go => taskemitter{,_test}.go because the new name is more proper after we merged the internal/task package inside of the oonimkall package; 2. rename runner.go's `run` function to `runTask`; 3. modify `runTask` to use the new `taskEmitterUsingChan` abstraction on which we will spend more works in a later point of this list; 4. introduce `runTaskWithEmitter` factory that is called by `runTask` and allows us to more easily write unit tests; 5. acknowledge that `runner` was not using its `out` field; 6. use the new `taskEmitterWrapper` in `newRunner`; 7. acknowledge that `runnerCallbacks` could use a generic `taskEmitter` as field type rather than a specific type; 8. rewrite tests to use `runTaskWithEmitter` which leads to simpler code that does not require a goroutine; 9. acknowledge that the code has been ignoring the `DisabledEvents` settings for quite some time, so stop supporting it; 10. refactor the `taskEmitter` implementation to be like: 1. we still have the `taskEmitter` interface; 2. `taskEmitterUsingChan` wraps the channel and allows for emitting events using the channel; 3. `taskEmitterUsingChan` owns an `eof` channel that is closed by `Close` (which is idempotent) and signals we should be stop emitting; 4. make sure `runTask` creates a `taskEmitterUsingChan` and calls its `Close` method when done; 5. completely remove the code for disabling events since the code was actually ignoring the stting; 6. add a `taskEmitterWrapper` that adds common functions for emitting events to _any_ `taskWrapper`; 7. write unit tests for `taskEmitterUsingChan` and for `taskEmitterWrapper`; 11. acknowledge that the abstraction we need for testing is actually a thread-safe thing that collects events into a vector containing events and refactor all tests accordingly. See https://github.com/ooni/probe/issues/1903 * [forwardport] refactor(oonimkall): make the runner unit-testable (#625) This diff forward ports 9423947faf6980d92d2fe67efe3829e8fef76586. See https://github.com/ooni/probe/issues/1903 * [forwardport] feat(oonimkall): write unit tests for the runner component (#626) This diff forward ports 35dd0e3788b8fa99c541452bbb5e0ae4871239e1. Forward porting note: compared to 35dd0e3788b8fa99c541452bbb5e0ae4871239e1, the diff I'm committing here is slightly different. In `master` we do not have the case where a measurement fails and a measurement is returned, thus I needed to adapt the test to become like this: ```diff diff --git a/pkg/oonimkall/runner_internal_test.go b/pkg/oonimkall/runner_internal_test.go index 334b574..84c7436 100644 --- a/pkg/oonimkall/runner_internal_test.go +++ b/pkg/oonimkall/runner_internal_test.go @@ -568,15 +568,6 @@ func TestTaskRunnerRun(t *testing.T) { }, { Key: failureMeasurement, Count: 1, - }, { - Key: measurement, - Count: 1, - }, { - Key: statusMeasurementSubmission, - Count: 1, - }, { - Key: statusMeasurementDone, - Count: 1, }, { Key: statusEnd, Count: 1, ``` I still need to write more assertions for each emitted event but the code we've here is already a great starting point. See https://github.com/ooni/probe/issues/1903 * [forwardport] refactor(oonimkall): merge files, use proper names, zap unneeded integration tests (#627) This diff forward ports f894427d24edc9a03fc78306d0093e7b51c46c25. Forward porting note: this diff is slightly different from the original mentioned above because it carries forward changes mentioned in the previous diff caused by a different way of handling a failed measurement in the master branch compared to the release/3.11 branch. Move everything that looked like "task's model" inside of the taskmodel.go file, for consistency. Make sure it's clear some variables are event types. Rename the concrete `runner` as `runnerForTask`. Also, remove now-unnecessary (and flaky!) integration tests for the `runnerForTask` type. While there, notice there were wrong URLs that were generated during the probe-engine => probe-cli move and fix them. See https://github.com/ooni/probe/issues/1903 * [forwardport] refactor(oonimkall): we can simplify StartTask tests (#628) This diff forward ports dcf2986c2032d8185d58d24130a7f2c2d61ef2fb. * refactor(oonimkall): we can simplify StartTask tests We have enough checks for runnerForTask. So we do not need to duplicate them when checking for StartTask. While there, refactor how we start tasks to remove the need for extra runner functions. This is the objective I wanted to achieve for oonimkall: 1. less duplicate tests, and 2. more unit tests (which are less flaky) At this point, we're basically done (pending forwardporting to master) with https://github.com/ooni/probe/issues/1903. * fix(oonimkall): TestStartTaskGood shouldn't cancel the test This creates a race condition where the test may fail if we cannot complete the whole "Example" test in less than one second. This should explain the build failures I've seen so far and why I didn't see those failures when running locally.
2021-12-02 12:47:07 +01:00
emitter := newTaskEmitterUsingChan(task.out)
r := newRunner(&settings, emitter)
r.Run(ctx)
[forwardport] fix(oonimkall): don't close channel to signal end of task (#619) (#620) This diff forward ports 90bf0629b957c912a5a6f3bb6c98ad3abb5a2ff6 to `master`. If we close the channel to signal the end of a task we may panic when some background goroutine tries to post on the channel. This bug is rare but may happen. See for example https://github.com/ooni/probe/issues/1438. How can we improve? First, let us add a timeout when sending to the channel. Given that the channel is buffered and we have a generous timeout (1/4 of a second), it's unlikely we will really block. But, in the event in which a late message appears, we'll eventually _unblock_ when sending with a timeout. So, now we don't have to worry anymore about leaking forever a goroutine. Then, let us change the protocol with which we signal that a task is done. We used to close the channel. Now, instead we just synchronously post a nil on the channel when done. In turn, we interpret this nil to mean that the task is done when we receive messages. The _main_ different with respect to before is that now we are asking the consumer of our API to drain the channel. Because before we had a blocking channel, it seems to me we were already requiring the consumer of the API to do that. Which means, I think in practical terms it did not change much. Finally, acknowledge that we don't need a specific state variable to tell us we're done and simplify a little bit the API by just making isRunning private and using the "we're done" signal to determine whether we've stopped running the task. All these changes should be enough to close https://github.com/ooni/probe/issues/1438.
2021-11-26 22:18:45 +01:00
task.out <- nil // signal that we're done w/o closing the channel
forwardport: pull the patches mentioned in ooni/probe#1908 (#629) * [forwardport] fix(oonimkall): make logger used by tasks unit testable (#623) This diff forward ports e4b04642c51e7461728b25941624e1b97ef0ec83. Reference issue: https://github.com/ooni/probe/issues/1903 * [forwardport] feat(oonimkall): improve taskEmitter testability (#624) This diff forward ports 3e0f01a389c1f4cdd7878ec151aff91870a0bdff. 1. rename eventemitter{,_test}.go => taskemitter{,_test}.go because the new name is more proper after we merged the internal/task package inside of the oonimkall package; 2. rename runner.go's `run` function to `runTask`; 3. modify `runTask` to use the new `taskEmitterUsingChan` abstraction on which we will spend more works in a later point of this list; 4. introduce `runTaskWithEmitter` factory that is called by `runTask` and allows us to more easily write unit tests; 5. acknowledge that `runner` was not using its `out` field; 6. use the new `taskEmitterWrapper` in `newRunner`; 7. acknowledge that `runnerCallbacks` could use a generic `taskEmitter` as field type rather than a specific type; 8. rewrite tests to use `runTaskWithEmitter` which leads to simpler code that does not require a goroutine; 9. acknowledge that the code has been ignoring the `DisabledEvents` settings for quite some time, so stop supporting it; 10. refactor the `taskEmitter` implementation to be like: 1. we still have the `taskEmitter` interface; 2. `taskEmitterUsingChan` wraps the channel and allows for emitting events using the channel; 3. `taskEmitterUsingChan` owns an `eof` channel that is closed by `Close` (which is idempotent) and signals we should be stop emitting; 4. make sure `runTask` creates a `taskEmitterUsingChan` and calls its `Close` method when done; 5. completely remove the code for disabling events since the code was actually ignoring the stting; 6. add a `taskEmitterWrapper` that adds common functions for emitting events to _any_ `taskWrapper`; 7. write unit tests for `taskEmitterUsingChan` and for `taskEmitterWrapper`; 11. acknowledge that the abstraction we need for testing is actually a thread-safe thing that collects events into a vector containing events and refactor all tests accordingly. See https://github.com/ooni/probe/issues/1903 * [forwardport] refactor(oonimkall): make the runner unit-testable (#625) This diff forward ports 9423947faf6980d92d2fe67efe3829e8fef76586. See https://github.com/ooni/probe/issues/1903 * [forwardport] feat(oonimkall): write unit tests for the runner component (#626) This diff forward ports 35dd0e3788b8fa99c541452bbb5e0ae4871239e1. Forward porting note: compared to 35dd0e3788b8fa99c541452bbb5e0ae4871239e1, the diff I'm committing here is slightly different. In `master` we do not have the case where a measurement fails and a measurement is returned, thus I needed to adapt the test to become like this: ```diff diff --git a/pkg/oonimkall/runner_internal_test.go b/pkg/oonimkall/runner_internal_test.go index 334b574..84c7436 100644 --- a/pkg/oonimkall/runner_internal_test.go +++ b/pkg/oonimkall/runner_internal_test.go @@ -568,15 +568,6 @@ func TestTaskRunnerRun(t *testing.T) { }, { Key: failureMeasurement, Count: 1, - }, { - Key: measurement, - Count: 1, - }, { - Key: statusMeasurementSubmission, - Count: 1, - }, { - Key: statusMeasurementDone, - Count: 1, }, { Key: statusEnd, Count: 1, ``` I still need to write more assertions for each emitted event but the code we've here is already a great starting point. See https://github.com/ooni/probe/issues/1903 * [forwardport] refactor(oonimkall): merge files, use proper names, zap unneeded integration tests (#627) This diff forward ports f894427d24edc9a03fc78306d0093e7b51c46c25. Forward porting note: this diff is slightly different from the original mentioned above because it carries forward changes mentioned in the previous diff caused by a different way of handling a failed measurement in the master branch compared to the release/3.11 branch. Move everything that looked like "task's model" inside of the taskmodel.go file, for consistency. Make sure it's clear some variables are event types. Rename the concrete `runner` as `runnerForTask`. Also, remove now-unnecessary (and flaky!) integration tests for the `runnerForTask` type. While there, notice there were wrong URLs that were generated during the probe-engine => probe-cli move and fix them. See https://github.com/ooni/probe/issues/1903 * [forwardport] refactor(oonimkall): we can simplify StartTask tests (#628) This diff forward ports dcf2986c2032d8185d58d24130a7f2c2d61ef2fb. * refactor(oonimkall): we can simplify StartTask tests We have enough checks for runnerForTask. So we do not need to duplicate them when checking for StartTask. While there, refactor how we start tasks to remove the need for extra runner functions. This is the objective I wanted to achieve for oonimkall: 1. less duplicate tests, and 2. more unit tests (which are less flaky) At this point, we're basically done (pending forwardporting to master) with https://github.com/ooni/probe/issues/1903. * fix(oonimkall): TestStartTaskGood shouldn't cancel the test This creates a race condition where the test may fail if we cannot complete the whole "Example" test in less than one second. This should explain the build failures I've seen so far and why I didn't see those failures when running locally.
2021-12-02 12:47:07 +01:00
emitter.Close()
fix(oonimkall): improve channel management pattern (#621) 1. add eof channel to event emitter and use this channel as signal that we shouldn't be sending anymore instead of using a pattern where we use a timer to decide sending has timed out (because we're using a buffered channel, it is still possible for some evetns to end up in the channel if there is space, but this is not a concern, because the events will be deleted when the channel itself is gone); 2. refactor all tests where we assumed the output channel was closed to actually use a parallel "eof" channel and use it as signal we should not be sending anymore (not strictly required but still the right thing to do in terms of using consistent patterns); 3. modify how we construct a runner so that it passes to the event emitter an "eof" channel and closes this channel when the main goroutine running the task is terminating; 4. modify the task to signal events such as "task goroutine started" and "task goroutine stopped" using channels, which helps to write much more correct tests; 5. take advantage of the previous change to improve the test that ensures we're not blocking for a small number of events and also improve the name of such a test to reflect what it's testing. The related issue in term of fixing the channel usage pattern is https://github.com/ooni/probe/issues/1438. Regarding improving testability, instead, the correct reference issue is https://github.com/ooni/probe/issues/1903. There are possibly more changes to apply here to improve this package and its testability, but let's land this diff first and then see how much time is left for further improvements. I've run unit and integration tests with `-race` locally. This diff will need to be backported to `release/3.11`.
2021-12-01 15:40:25 +01:00
close(task.isstopped)
}()
return task, nil
}
// WaitForNextEvent blocks until the next event occurs. The returned
// string is a serialized JSON following MK v0.10.9's API.
func (t *Task) WaitForNextEvent() string {
const terminated = `{"key":"task_terminated","value":{}}` // like MK
[forwardport] fix(oonimkall): don't close channel to signal end of task (#619) (#620) This diff forward ports 90bf0629b957c912a5a6f3bb6c98ad3abb5a2ff6 to `master`. If we close the channel to signal the end of a task we may panic when some background goroutine tries to post on the channel. This bug is rare but may happen. See for example https://github.com/ooni/probe/issues/1438. How can we improve? First, let us add a timeout when sending to the channel. Given that the channel is buffered and we have a generous timeout (1/4 of a second), it's unlikely we will really block. But, in the event in which a late message appears, we'll eventually _unblock_ when sending with a timeout. So, now we don't have to worry anymore about leaking forever a goroutine. Then, let us change the protocol with which we signal that a task is done. We used to close the channel. Now, instead we just synchronously post a nil on the channel when done. In turn, we interpret this nil to mean that the task is done when we receive messages. The _main_ different with respect to before is that now we are asking the consumer of our API to drain the channel. Because before we had a blocking channel, it seems to me we were already requiring the consumer of the API to do that. Which means, I think in practical terms it did not change much. Finally, acknowledge that we don't need a specific state variable to tell us we're done and simplify a little bit the API by just making isRunning private and using the "we're done" signal to determine whether we've stopped running the task. All these changes should be enough to close https://github.com/ooni/probe/issues/1438.
2021-11-26 22:18:45 +01:00
if t.isdone.Load() != 0 {
return terminated
}
evp := <-t.out
if evp == nil {
t.isdone.Add(1)
return terminated
}
data, err := json.Marshal(evp)
runtimex.PanicOnError(err, "json.Marshal failed")
return string(data)
}
// IsDone returns true if the task is done.
func (t *Task) IsDone() bool {
return t.isdone.Load() != 0
}
// Interrupt interrupts the task.
func (t *Task) Interrupt() {
t.cancel()
}