-
Notifications
You must be signed in to change notification settings - Fork 63
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #2394 from lf-lang/dataflow
Improvements in decentralized coordination
- Loading branch information
Showing
14 changed files
with
239 additions
and
69 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Submodule reactor-c
updated
9 files
| +3 −0 | core/federated/RTI/.gitignore | |
| +77 −23 | core/federated/federate.c | |
| +0 −2 | core/reactor_common.c | |
| +18 −57 | core/threaded/reactor_threaded.c | |
| +1 −1 | core/threaded/scheduler_NP.c | |
| +8 −6 | core/threaded/scheduler_sync_tag_advance.c | |
| +15 −5 | include/core/federated/federate.h | |
| +29 −1 | include/core/threaded/reactor_threaded.h | |
| +1 −1 | lingua-franca-ref.txt |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,64 @@ | ||
| target C { | ||
| coordination: decentralized // logging: debug | ||
| } | ||
|
|
||
| reactor Client(STP_offset: time = 1 day) { | ||
| input server_message: int | ||
| output client_message: int | ||
|
|
||
| reaction(startup) {= | ||
| lf_print("Client Startup!"); | ||
| =} | ||
|
|
||
| reaction(server_message) -> client_message {= | ||
| int val = server_message->value + 1; | ||
| lf_sleep(MSEC(100)); | ||
| lf_print("client: %d", val); | ||
| if (val == 9) { | ||
| lf_print("client requesting stop"); | ||
| lf_request_stop(); | ||
| } | ||
| lf_set(client_message, val); | ||
| =} STP(0) {= | ||
| // Zero STAA because STA is large and gets added. | ||
| lf_print_error_and_exit("Client STP Violated!"); | ||
| =} | ||
| } | ||
|
|
||
| reactor Server(STP_offset: time = 1 day) { | ||
| output server_message: int | ||
| input client_message1: int | ||
| input client_message2: int | ||
|
|
||
| reaction(startup) -> server_message {= | ||
| lf_print("Server Startup!"); | ||
| lf_set(server_message, 0); | ||
| =} | ||
|
|
||
| reaction(client_message1, client_message2) -> server_message {= | ||
| int val = client_message1->value; | ||
| if (val < client_message2->value) val = client_message2->value; | ||
| lf_sleep(MSEC(100)); | ||
| val += 1; | ||
| lf_print("server: %d", val); | ||
| if (val == 8) { | ||
| lf_print("server requesting stop"); | ||
| lf_set(server_message, val); | ||
| lf_request_stop(); | ||
| } | ||
| lf_set(server_message, val); | ||
| =} STP(0) {= | ||
| // Zero STAA because STA is large and gets added. | ||
| lf_print_error_and_exit("Server STP Violated!"); | ||
| =} | ||
| } | ||
|
|
||
| federated reactor { | ||
| client1 = new Client() | ||
| client2 = new Client() | ||
| server = new Server() | ||
| server.server_message -> client1.server_message | ||
| client1.client_message -> server.client_message1 after 0 | ||
| server.server_message -> client2.server_message | ||
| client2.client_message -> server.client_message2 after 0 | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| /** | ||
| * @brief A federated system with a decentralized coordinator that has large STA and STAA offsets. | ||
| * | ||
| * This test verifies that a large STA and STAA offset can be used when the data flow is predictable | ||
| * even if the the program is lagging behind physical time. | ||
| * | ||
| * @author Edward A. Lee | ||
| */ | ||
| target C { | ||
| coordination: decentralized, | ||
| timeout: 40 ns | ||
| } | ||
|
|
||
| // Pi needs an STP_offset because its event queue is usually empty and otherwise it will advance to the stop time. | ||
| reactor Pi(STP_offset: time = 1 day) { | ||
| input trigger: bool | ||
| output out: int | ||
|
|
||
| reaction(trigger) -> out {= | ||
| tag_t now = lf_tag(); | ||
| lf_print("***** at tag " PRINTF_TAG, now.time - lf_time_start(), now.microstep); | ||
| lf_set(out, 42); | ||
| =} STP(30 s) {= | ||
| tag_t now = lf_tag(); | ||
| lf_print_error_and_exit("STP violation at Pi at tag " PRINTF_TAG, now.time - lf_time_start(), now.microstep); | ||
| =} | ||
| } | ||
|
|
||
| // Gather doesn't need an STP_offset because its event queue is never empty. | ||
| reactor Gather { | ||
| input[4] in: int | ||
| output next: bool | ||
| logical action a(10 ns) | ||
| state count: int = 0 | ||
|
|
||
| reaction(startup, a) -> next {= | ||
| lf_set(next, true); | ||
| =} | ||
|
|
||
| reaction(in) -> a {= | ||
| tag_t now = lf_tag(); | ||
| for (int i = 0; i < 4; i++) { | ||
| if (!in[i]->is_present) { | ||
| lf_print_error_and_exit("Missing input %d in Gather at tag " PRINTF_TAG, | ||
| i, now.time - lf_time_start(), now.microstep); | ||
| } | ||
| } | ||
| lf_print("%d: at tag " PRINTF_TAG, self->count, now.time - lf_time_start(), now.microstep); | ||
| self->count++; | ||
| lf_schedule(a, 0); | ||
| =} STP(1 day) {= | ||
| tag_t now = lf_tag(); | ||
| lf_print_error_and_exit("STP violation at Gather at tag " PRINTF_TAG, now.time - lf_time_start(), now.microstep); | ||
| =} | ||
|
|
||
| reaction(shutdown) {= | ||
| if (self->count < 5) { | ||
| lf_print_error_and_exit("Gather received only %d inputs. Expected 5.", self->count); | ||
| } | ||
| =} | ||
| } | ||
|
|
||
| federated reactor { | ||
| pi = new[4] Pi() | ||
| g = new Gather() | ||
| pi.out -> g.in | ||
| (g.next)+ -> pi.trigger | ||
| } |
Oops, something went wrong.