diff --git a/.gitignore b/.gitignore index 4d0ba53..93e76c3 100644 --- a/.gitignore +++ b/.gitignore @@ -2,4 +2,4 @@ qs_*.* bak bin/ -*.exe +*~ \ No newline at end of file diff --git a/README.md b/README.md index 1becf53..3a606d8 100644 --- a/README.md +++ b/README.md @@ -1,9 +1,11 @@ # GoFBP -This repo holds the beginning of an FBP implementation in Go +This repo contains an FBP implementation in Go. It conforms pretty closely to the scheduling logic in JavaFBP, C#FBP and C++FBP (all on https://github.com/jpaulm ). There may well be further internal changes, but I am hoping that the "external" APIs (network and component definitions) are now firm. +Current tag: **v1.0.6** . + ## General General web site for "classical" FBP: @@ -12,8 +14,10 @@ General web site for "classical" FBP: In computer programming, flow-based programming (FBP) is a programming paradigm that defines applications as networks of "black box" processes, which exchange data across predefined connections by message passing, where the connections are specified externally to the processes. These black box processes can be reconnected endlessly to form different applications without having to be changed internally. FBP is thus naturally component-oriented. FBP is a particular form of dataflow programming based on bounded buffers, information packets with defined lifetimes, named ports, and separate definition of connections. + +DrawFBP (https://github.com/jpaulm/drawfbp) can now generate working GoFBP network and subnet definitions from a network diagram (as well as JavaFBP, C#FBP, JSON and free-form notations). -GoFBP Network Definition Syntax and Component API: +## GoFBP Network Definition Syntax and Component API: * https://jpaulm.github.io/fbp/gosyntax.htm ## Features (these are common to all FBP implementations on GitHub/jpaulm): @@ -25,19 +29,62 @@ GoFBP Network Definition Syntax and Component API: - "automatic" in- and out-ports - notation is port name = "*" - GoFBP (and the other FBP implementations on https://github.com/jpaulm distingish between "invocation" and "activation" of processes: a process is invoked once, but may be activated multiple times (if it does a return before all its input ports have been drained) +## Running your app with GoFBP + +In the `go.mod` file in your root, add the statement + +``` +require github.com/jpaulm/gofbp latest +``` +and then run the command `go mod tidy` - this will change the word `latest` to the latest version, and store it back in your `go.mod` file. + +If you need parameter values other than `false`, you will have to access a `params.xml` file, as shown below. + +Command to run your network, e.g.: + +``` +go run Merge.go +``` + +A number of test cases are in the `testing` folder, and can be run as follows: + +``` +cd testing +go test +``` +To run an individual network, e.g. `TestIntSender` in `testing\testintsender_test.go`, +run +``` +cd testing +go test -run IntSender +``` + +### Notes: + +- One comment about running with subnets: the subnet or subnets should be in a different folder from the code that invokes it. If they are in the same folder, apparently both the invoking code and the subnet need to be named individually in a `go run` command... this is not required when running under VSCode, or from an `.exe` file. + ## Tracing -An XML file has been provided in the root, called `params.xml`. So far there is only one parameter: +An XML file has been provided in the root, called `params.xml`. If you need values other than all `false`, you will need to reference this using the `LoadXMLParams` and `SetParams` methods - see https://jpaulm.github.io/fbp/gosyntax.htm . -
-<?xml version="1.0"?>
-<tracing>true|false</tracing>
-<tracelocks>true|false</tracelocks>
-
+Format of the tracing definitions file: + +``` + + +false +false +false (traces packet creates, createbrackets and discards) +true + +``` +The `generate-gIds` parameter is only used to assist in debugging deadlocks, so can default most of the time - see below. ## Subnets These are described in Chap. 7 in "Flow-Based Programming": Composite Components - https://jpaulmorrison.com/fbp/compos.shtml , although we haven't implemented dynamic subnets (yet)! +**Note:** If a subnet `.go` file is in the same folder as the network `.go` file invoking it, the `NewProc` call should be written without the folder - e.g. `net.NewProc("Run___Subnet", &Subnet1{})`, but currently DrawFBP cannot detect this situation, so, when using notation `GoFBP`, this call will not be generated correctly. + ## Test Cases The following test cases are now working - thanks to Egon Elbre and Emil Valeev for all their help! @@ -56,9 +103,11 @@ The following test cases are now working - thanks to Egon Elbre and Emil Valeev - test subnet (SubIn and SubOut) - force deadlock (separate test file) - this is designed to crash, and in fact will give a message if it does *not* crash! + +- simple web sockets test -To run them, position to your `GitHub\gofbp` directory, and do any of the following: +To run them, position to your `GitHub\gofbp\testing` directory (note the additional folder, as of tag `v1.0.3` plus), and do any of the following: - `go test -run Merge -count=1` - `go test -run Concat -count=1` @@ -69,9 +118,11 @@ To run them, position to your `GitHub\gofbp` directory, and do any of the follow - `go test -run WriteToConsUsingNL -count=1` (note the activated/deactivated messages) - `go test -run ForceDeadlock -count=1` - `go test -run InfQueueAsMain -count=1` (note the "automatic" ports between WriteFile and ReadFile) +- `go test -run LoadBal -count=1` (this does load balancing, and uses two DelayedReceiver processes) - `go test -run Subnet1 -count=1` - `go test -run Subnet2 -count=1` - `go test -run Subnet3 -count=1` +- `go test -run TestWebSocket -count=1` - for more instructions, see # Test WebSockets (below) **Note**: ForceDeadlock is constructed differently so that it can "crash" without disrupting the flow of tests: the network definition has to be compiled "on the fly", so it is actually in `testdata`, while the test itself contains the code to compile and run the test. @@ -130,14 +181,34 @@ The following components are available: - `readfile.go` - `writefile.go` +"websocket" folder (for instructions, see below): +- `ws_request.go` +- `ws_respond.go` +- `ws_ans_req.go` (sample component - properly belongs in test suite) + +**Test Websockets** + +- position to your `GitHub\gofbp\testing` directory in DOS +- run `go test -run TestWebSocket -count=1` + +- in File Explorer, locate `GitHub\gofbp\scripts\chat2.html` +- open with Firefox or Chrome +- enter `namelist` in the Command box, hit enter or Send +- you should see `Server: Line1`, `Server: Line2`, `Server: Line3` show up in the box below "Send" +- you can repeat this multiple times, and the 3 output lines will be appended each time to the current display +- click on `Stop WS` - you will see `End of dialog` next to `Status` + +- you're done... except that, right now, it's not coming down cleanly, so you will have to close the DOS session manually. Hopefully, this will be fixed soon! + + **To dos** -- More and better documentation - Convert `panic`s to more standard Go error handling -- Way too much logging - have to make that optional - put remaining logging under switch control +- Way too much logging - have to make that optional - put remaining logging under switch control - *done!* - Add subnet handling - *done!* -- Generate GoFBP networks from DrawFBP - https://github.com/jpaulm/drawfbp -- Add Load Balancing component -- Add sample code showing use of substreams +- Generate GoFBP networks from DrawFBP - https://github.com/jpaulm/drawfbp - *done!* +- Add Load Balancing component - *done!* +- Add sample code showing use of substreams - *done!* - "Automatic" ports - *done!* -- Add Lua interface - see https://jpaulm.github.io/fbp/thlua.html +- Add Lua interface - similar to https://jpaulm.github.io/fbp/thlua.html + diff --git a/analyze_deadlock_compile.bat b/analyze_deadlock_compile.bat new file mode 100644 index 0000000..b921a0d --- /dev/null +++ b/analyze_deadlock_compile.bat @@ -0,0 +1 @@ +go build -o analyze_deadlock.exe utils\analyze_deadlock.go \ No newline at end of file diff --git a/cmd/merge.go b/cmd/merge.go new file mode 100644 index 0000000..7e58350 --- /dev/null +++ b/cmd/merge.go @@ -0,0 +1,25 @@ +package main //change package name, or delete statement, if desired + +// Network generated by DrawFBP + +import ( + "github.com/jpaulm/gofbp/components/testrtn" + "github.com/jpaulm/gofbp/core" +) + +func main() { + params, err := core.LoadXMLParams("params.xml") + if err != nil { + panic(err) + } + net := core.NewNetwork("Merge") + net.SetParams(params) + sender2 := net.NewProc("Sender2", &testrtn.Sender{}) + write___to_console := net.NewProc("Write___To_Console", &testrtn.WriteToConsole{}) + sender1 := net.NewProc("Sender1", &testrtn.Sender{}) + net.Connect(sender1, "OUT", write___to_console, "IN", 6) + net.Initialize("15", sender1, "COUNT") + net.Initialize("10", sender2, "COUNT") + net.Connect(sender2, "OUT", write___to_console, "IN", 6) + net.Run() +} diff --git a/component.go b/component.go new file mode 100644 index 0000000..d75ac47 --- /dev/null +++ b/component.go @@ -0,0 +1,11 @@ +package gofbp + +type Component interface { + Setup(*Process) + Execute(*Process) +} + +type ComponentWithMustRun interface { + Component + MustRun() bool +} diff --git a/components/io/readfile.go b/components/io/readfile.go index c63d70f..20bb3e7 100644 --- a/components/io/readfile.go +++ b/components/io/readfile.go @@ -1,3 +1,4 @@ +/*Package io implements gofbp I/O*/ package io import ( @@ -10,16 +11,19 @@ import ( "github.com/jpaulm/gofbp/core" ) +//ReadFile type defines ipt and opt type ReadFile struct { ipt core.InputConn opt core.OutputConn } +//Setup method opens readFile func (readFile *ReadFile) Setup(p *core.Process) { readFile.ipt = p.OpenInPort("FILENAME") readFile.opt = p.OpenOutPort("OUT") } +//Execute method starts Process func (readFile *ReadFile) Execute(p *core.Process) { icpkt := p.Receive(readFile.ipt) diff --git a/components/io/writefile.go b/components/io/writefile.go index 5a420a3..a9d34c2 100644 --- a/components/io/writefile.go +++ b/components/io/writefile.go @@ -7,34 +7,39 @@ import ( "github.com/jpaulm/gofbp/core" ) +// WriteFile type defines iptIP, ipt, and opt type WriteFile struct { - iptIp core.InputConn + iptIP core.InputConn ipt core.InputConn opt core.OutputConn } +//Setup method initializes Process func (writeFile *WriteFile) Setup(p *core.Process) { - writeFile.iptIp = p.OpenInPort("FILENAME") + writeFile.iptIP = p.OpenInPort("FILENAME") writeFile.ipt = p.OpenInPort("IN") writeFile.opt = p.OpenOutPortOptional("OUT") } +//MustRun method func (WriteFile) MustRun() {} +//Execute method starts Process func (writeFile *WriteFile) Execute(p *core.Process) { - icpkt := p.Receive(writeFile.iptIp) + icpkt := p.Receive(writeFile.iptIP) fname, ok := icpkt.Contents.(string) if !ok { panic("Parameter (file name) not a string") } p.Discard(icpkt) - p.Close(writeFile.iptIp) + p.Close(writeFile.iptIP) f, err := os.Create(fname) if err != nil { panic("Unable to open file: " + fname) } + defer fmt.Println(p.Name+": File", fname, "written") defer f.Close() for { @@ -43,14 +48,14 @@ func (writeFile *WriteFile) Execute(p *core.Process) { break } - data := []byte(pkt.Contents.(string) + "\n") - - _, err2 := f.Write(data) + data := fmt.Sprint(pkt.Contents) + _, err2 := fmt.Fprintln(f, data) if err2 != nil { panic("Unable to write file: " + fname) } + f.Sync() if !writeFile.opt.IsConnected() { p.Discard(pkt) } else { @@ -58,5 +63,4 @@ func (writeFile *WriteFile) Execute(p *core.Process) { } } - fmt.Println(p.Name+": File", fname, "written") } diff --git a/components/subnets/sssubnet1.go b/components/subnets/sssubnet1.go index db07317..e979044 100644 --- a/components/subnets/sssubnet1.go +++ b/components/subnets/sssubnet1.go @@ -1,3 +1,4 @@ +/*Package subnets implements FBP subnet processing*/ package subnets import ( diff --git a/components/subnets/subnet1.go b/components/subnets/subnet1.go index 7147336..39182d0 100644 --- a/components/subnets/subnet1.go +++ b/components/subnets/subnet1.go @@ -25,3 +25,5 @@ func (subnet *Subnet1) Execute(p *core.Process) { net.Run() } + +// NAME->0(proc1 SubIn)1.OUT -> IN.0(Proc2 WriteToConsole1)1.OUT -> NAME.0(proc3 SubOut); diff --git a/components/testrtn/concatstr.go b/components/testrtn/concatstr.go index 39dbf07..b491bc7 100644 --- a/components/testrtn/concatstr.go +++ b/components/testrtn/concatstr.go @@ -1,6 +1,8 @@ package testrtn -import "github.com/jpaulm/gofbp/core" +import ( + "github.com/jpaulm/gofbp/core" +) type ConcatStr struct { ipt core.InputArrayConn diff --git a/components/testrtn/counter.go b/components/testrtn/counter.go index 1b2134d..aea187e 100644 --- a/components/testrtn/counter.go +++ b/components/testrtn/counter.go @@ -22,7 +22,6 @@ func (counter *Counter) Setup(p *core.Process) { func (Counter) MustRun() {} func (counter *Counter) Execute(p *core.Process) { - count := 0 @@ -43,5 +42,4 @@ func (counter *Counter) Execute(p *core.Process) { pkt := p.Create(strconv.Itoa(count)) p.Send(counter.cnt, pkt) - } diff --git a/components/testrtn/discard.go b/components/testrtn/discard.go index 3cd3c6d..d77e585 100644 --- a/components/testrtn/discard.go +++ b/components/testrtn/discard.go @@ -17,7 +17,6 @@ func (discard *Discard) Setup(p *core.Process) { //func (Discard) MustRun() {} func (discard *Discard) Execute(p *core.Process) { - for { var pkt = p.Receive(discard.ipt) @@ -30,5 +29,4 @@ func (discard *Discard) Execute(p *core.Process) { } - } diff --git a/components/testrtn/intsender.go b/components/testrtn/intsender.go new file mode 100644 index 0000000..dd352a9 --- /dev/null +++ b/components/testrtn/intsender.go @@ -0,0 +1,37 @@ +/*Package testrtn tests gofbp code.*/ +package testrtn + +import ( + //"fmt" + "strconv" + + "github.com/jpaulm/gofbp/core" +) + +/*Sender type defines ipt and opt for process send*/ +type IntSender struct { + ipt core.InputConn + opt core.OutputConn +} + +/*Setup function initializes a source process.*/ +func (intsender *IntSender) Setup(p *core.Process) { + intsender.ipt = p.OpenInPort("COUNT") + intsender.opt = p.OpenOutPort("OUT") +} + +/*Execute function launches a source process.*/ +func (intsender *IntSender) Execute(p *core.Process) { + + icpkt := p.Receive(intsender.ipt) + j, _ := strconv.Atoi(icpkt.Contents.(string)) + p.Discard(icpkt) + p.Close(intsender.ipt) + + var pkt *core.Packet + for i := 0; i < j; i++ { + pkt = p.Create(i) + p.Send(intsender.opt, pkt) + } + +} diff --git a/components/testrtn/kick.go b/components/testrtn/kick.go index ae05160..e9db4af 100644 --- a/components/testrtn/kick.go +++ b/components/testrtn/kick.go @@ -15,10 +15,8 @@ func (kick *Kick) Setup(p *core.Process) { } func (kick *Kick) Execute(p *core.Process) { - var pkt = p.Create("Kicker IP") p.Send(kick.opt, pkt) - } diff --git a/components/testrtn/prefix.go b/components/testrtn/prefix.go index 261fde2..8397f37 100644 --- a/components/testrtn/prefix.go +++ b/components/testrtn/prefix.go @@ -8,7 +8,7 @@ import ( type Prefix struct { ipt core.InputConn - iptIp core.InputConn + iptIP core.InputConn out core.OutputConn } @@ -18,17 +18,17 @@ func (prefix *Prefix) Setup(p *core.Process) { prefix.out = p.OpenOutPort("OUT") - prefix.iptIp = p.OpenInPort("PARAM") + prefix.iptIP = p.OpenInPort("PARAM") } func (prefix *Prefix) Execute(p *core.Process) { - icpkt := p.Receive(prefix.iptIp) + icpkt := p.Receive(prefix.iptIP) param := icpkt.Contents.(string) p.Discard(icpkt) - p.Close(prefix.iptIp) + p.Close(prefix.iptIP) for { var pkt = p.Receive(prefix.ipt) diff --git a/components/testrtn/selector.go b/components/testrtn/selector.go index d3adc87..85b2987 100644 --- a/components/testrtn/selector.go +++ b/components/testrtn/selector.go @@ -9,7 +9,7 @@ import ( type Selector struct { ipt core.InputConn - iptIp core.InputConn + iptIP core.InputConn out1 core.OutputConn out2 core.OutputConn } @@ -22,20 +22,20 @@ func (selector *Selector) Setup(p *core.Process) { selector.out2 = p.OpenOutPortOptional("REJ") - selector.iptIp = p.OpenInPort("PARAM") + selector.iptIP = p.OpenInPort("PARAM") } func (Selector) MustRun() {} func (selector *Selector) Execute(p *core.Process) { - icpkt := p.Receive(selector.iptIp) + icpkt := p.Receive(selector.iptIP) param := icpkt.Contents.(string) i := len(param) p.Discard(icpkt) - p.Close(selector.iptIp) + p.Close(selector.iptIP) for { var pkt = p.Receive(selector.ipt) diff --git a/components/testrtn/sender.go b/components/testrtn/sender.go index b73d262..751da1b 100644 --- a/components/testrtn/sender.go +++ b/components/testrtn/sender.go @@ -1,3 +1,4 @@ +/*Package testrtn tests gofbp code.*/ package testrtn import ( @@ -7,18 +8,21 @@ import ( "github.com/jpaulm/gofbp/core" ) +/*Sender type defines ipt and opt for process send*/ type Sender struct { ipt core.InputConn opt core.OutputConn } +/*Setup function initializes a source process.*/ func (sender *Sender) Setup(p *core.Process) { sender.ipt = p.OpenInPort("COUNT") sender.opt = p.OpenOutPort("OUT") } +/*Execute function launches a source process.*/ func (sender *Sender) Execute(p *core.Process) { - + icpkt := p.Receive(sender.ipt) j, _ := strconv.Atoi(icpkt.Contents.(string)) p.Discard(icpkt) @@ -29,5 +33,5 @@ func (sender *Sender) Execute(p *core.Process) { pkt = p.Create("IP - # " + strconv.Itoa(i)) p.Send(sender.opt, pkt) } - + } diff --git a/components/testrtn/writetoconsnl.go b/components/testrtn/writetoconsnl.go index 435ae7b..0d1999a 100644 --- a/components/testrtn/writetoconsnl.go +++ b/components/testrtn/writetoconsnl.go @@ -29,12 +29,12 @@ func (writeToConsole *WriteToConsNL) Execute(p *core.Process) { return } if pkt.PktType == core.OpenBracket { - fmt.Println("Open", pkt.Contents) + fmt.Println("Open Bracket", pkt.Contents) } else { if pkt.PktType == core.CloseBracket { - fmt.Println("Close", pkt.Contents) + fmt.Println("Close Bracket", pkt.Contents) } else { - fmt.Println(pkt.Contents) + fmt.Print(pkt.Contents) } } diff --git a/components/testrtn/writetoconsole.go b/components/testrtn/writetoconsole.go index 3140b87..8436478 100644 --- a/components/testrtn/writetoconsole.go +++ b/components/testrtn/writetoconsole.go @@ -26,12 +26,12 @@ func (writeToConsole *WriteToConsole) Execute(p *core.Process) { break } if pkt.PktType == core.OpenBracket { - fmt.Println("Open", pkt.Contents) + fmt.Println("Open Bracket", pkt.Contents) } else { if pkt.PktType == core.CloseBracket { - fmt.Println("Close", pkt.Contents) + fmt.Println("Close Bracket", pkt.Contents) } else { - fmt.Println(pkt.Contents) + fmt.Print(pkt.Contents) } } diff --git a/components/websocket/ans/params.xml b/components/websocket/ans/params.xml new file mode 100644 index 0000000..04eba75 --- /dev/null +++ b/components/websocket/ans/params.xml @@ -0,0 +1,6 @@ + + + true + false + false + diff --git a/components/websocket/ans/ws_ans_req.go b/components/websocket/ans/ws_ans_req.go new file mode 100644 index 0000000..27ad48f --- /dev/null +++ b/components/websocket/ans/ws_ans_req.go @@ -0,0 +1,49 @@ +package ans + +import ( + "github.com/jpaulm/gofbp/core" +) + +// WSAnsReq struct contains input and output gofbp.core socket connections +type WSAnsReq struct { + ipt core.InputConn + out core.OutputConn +} + +// Setup Function defines in and out ports +func (wsansreq *WSAnsReq) Setup(p *core.Process) { + wsansreq.ipt = p.OpenInPort("IN") + wsansreq.out = p.OpenOutPortOptional("OUT") +} + +// MustRun Function is defined elsewhere +func (WSAnsReq) MustRun() {} + +// Execute Function runs a WSAnsReq gofbp component instance process +func (wsansreq *WSAnsReq) Execute(p *core.Process) { + + for { + pkt := p.Receive(wsansreq.ipt) // open bracket + if pkt == nil { + break + } + p.Send(wsansreq.out, pkt) + + pkt = p.Receive(wsansreq.ipt) // connection + p.Send(wsansreq.out, pkt) + + pkt = p.Receive(wsansreq.ipt) //"namelist" + p.Discard(pkt) + + pkt = p.Create("line1") + p.Send(wsansreq.out, pkt) + pkt = p.Create("line2") + p.Send(wsansreq.out, pkt) + pkt = p.Create("line3") + p.Send(wsansreq.out, pkt) + + pkt = p.Receive(wsansreq.ipt) // close bracket + p.Send(wsansreq.out, pkt) + } + +} diff --git a/components/websocket/ans/ws_ans_req_test.go b/components/websocket/ans/ws_ans_req_test.go new file mode 100644 index 0000000..22f6a37 --- /dev/null +++ b/components/websocket/ans/ws_ans_req_test.go @@ -0,0 +1,27 @@ +// package ans (ws_ans_req.go) is a totally .... intended just to fill out the structure of a GoFBP web socket server... +package ans + +import ( + "testing" + + "github.com/jpaulm/gofbp/components/websocket" + "github.com/jpaulm/gofbp/core" +) + +func TestWebSocket(t *testing.T) { + params, err := core.LoadXMLParams("params.xml") + if err != nil { + panic(err) + } + path := "localhost:8080" + net := core.NewNetwork("TestWebSocket") + net.SetParams(params) + //proc1 := net.NewProc("WSRequest", &websocket.WSRequest{}) + proc1 := net.NewProc("WSRequest", &websocket.WSRequest{}) // Assumed to be working + proc2 := net.NewProc("AnsReq", &WSAnsReq{}) // Testing this one. + proc3 := net.NewProc("WSRespond", &websocket.WSRespond{}) // Assumed to be working + net.Initialize(path, proc1, "ADDR") + net.Connect(proc1, "OUT", proc2, "IN", 6) + net.Connect(proc2, "OUT", proc3, "IN", 6) + net.Run() +} diff --git a/components/websocket/ws_request.go b/components/websocket/ws_request.go new file mode 100644 index 0000000..2316f75 --- /dev/null +++ b/components/websocket/ws_request.go @@ -0,0 +1,155 @@ +// https://github.com/gorilla/websocket/blob/master/examples/echo/server.go + +// try https://stackoverflow.com/questions/39320025/how-to-stop-http-listenandserve + +package websocket + +import ( + "context" + "fmt" + "log" + "net/http" + "sync" + "time" + + "github.com/gorilla/websocket" + "github.com/jpaulm/gofbp/core" +) + +//var data_map map[*websocket.Conn][]*core.Packet + +type WSRequest struct { + ipt core.InputConn + opt core.OutputConn +} + +func (wsrequest *WSRequest) Setup(p *core.Process) { + wsrequest.ipt = p.OpenInPort("ADDR") + wsrequest.opt = p.OpenOutPort("OUT") +} + +var upgrader = websocket.Upgrader{} +var proc *core.Process + +func (wsrequest *WSRequest) Execute(p *core.Process) { + icpkt := p.Receive(wsrequest.ipt) + path := icpkt.Contents.(string) + p.Discard(icpkt) + p.Close(wsrequest.ipt) + proc = p + + log.SetFlags(0) + + log.Printf("main: starting HTTP server") + + httpServerExitDone := &sync.WaitGroup{} + + httpServerExitDone.Add(1) + + srv := startHttpServer(httpServerExitDone, path) + + log.Printf("serving for 20 seconds") + + time.Sleep(20 * time.Second) + + log.Printf("stopping HTTP server") + + // now close the server gracefully ("shutdown") + // timeout could be given with a proper context + // (in real world you shouldn't use TODO()). ???????? + if err := srv.Shutdown(context.TODO()); err != nil { + panic(err) // failure/timeout shutting down the server gracefully + } + + // wait for goroutine started in startHttpServer() to stop + httpServerExitDone.Wait() + + log.Printf("done. exiting") + +} + +func startHttpServer(wg *sync.WaitGroup, path string) *http.Server { + srv := &http.Server{Addr: path} + + //http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + http.HandleFunc("/", serveWs) + + go func() { + defer wg.Done() // let main know we are done cleaning up + + // always returns error. ErrServerClosed on graceful close + if err := srv.ListenAndServe(); err != http.ErrServerClosed { + // unexpected error. port in use? + log.Fatalf("ListenAndServe(): %v", err) + } + }() + + // returning reference so caller can call Shutdown() + return srv +} + +func serveWs(w http.ResponseWriter, r *http.Request) { + fmt.Println("upgrade") + upgrader.CheckOrigin = func(r *http.Request) bool { return true } + c, err := upgrader.Upgrade(w, r, nil) + if err != nil { + log.Print("upgrade:", err) + return + } + //conn = *c + defer c.Close() + + var pkt_list []*core.Packet + var pkt *core.Packet + + opt := proc.OpenOutPort("OUT") + + for { + _, message, err := c.ReadMessage() + if err != nil { + log.Println("read:", err) + break + } + log.Printf("recv: %s", message) + + x := string(message) + + if x == "@{" { + + pkt_list = make([]*core.Packet, 0) + + continue + } + + if x == "@}" { + + // send out "connection" IPs, then IPs stored in pkt_list ... all surrounded by bracket IPs + + pkt = proc.CreateBracket(core.OpenBracket, "") + proc.Send(opt, pkt) + + pkt = proc.Create(c) // connection + proc.Send(opt, pkt) + + for _, pkt := range pkt_list { + proc.Send(opt, pkt) + } + + //data_map[c] = nil + pkt = proc.CreateBracket(core.CloseBracket, "") + proc.Send(opt, pkt) + continue + } + + if x == "@kill" { + + c.Close() + //break + continue + } + + pkt = proc.Create(x) + pkt_list = append(pkt_list, pkt) + + } +} diff --git a/components/websocket/ws_respond.go b/components/websocket/ws_respond.go new file mode 100644 index 0000000..e44b2d1 --- /dev/null +++ b/components/websocket/ws_respond.go @@ -0,0 +1,55 @@ +// https://github.com/gorilla/websocket/blob/master/examples/echo/server.go + +package websocket + +import ( + "log" + + "github.com/gorilla/websocket" + "github.com/jpaulm/gofbp/core" +) + +type WSRespond struct { + ipt core.InputConn +} + +func (WSRespond) MustRun() {} + +func (wsrespond *WSRespond) Setup(p *core.Process) { + wsrespond.ipt = p.OpenInPort("IN") +} + +func (wsrespond *WSRespond) Execute(p *core.Process) { + for { + pkt := p.Receive(wsrespond.ipt) // should be open bracket + if pkt == nil { + return + } + p.Discard(pkt) + + pkt = p.Receive(wsrespond.ipt) // connection + if pkt == nil { + return + } + conn, _ := pkt.Contents.(*websocket.Conn) + + p.Discard(pkt) + + for { + pkt = p.Receive(wsrespond.ipt) + if pkt.PktType == core.CloseBracket { + break + } + + err := conn.WriteMessage(websocket.TextMessage, []byte(pkt.Contents.(string))) + if err != nil { + log.Println("write:", err) + break + } else { + p.Discard(pkt) + } + } + p.Discard(pkt) + } + +} diff --git a/core/component.go b/core/component.go index d7d96d6..0aa2e97 100644 --- a/core/component.go +++ b/core/component.go @@ -1,10 +1,13 @@ +/*Package core implements gofbp's run time engine.*/ package core +/*The Component interface implements component Setup and Execute methods.*/ type Component interface { Setup(*Process) Execute(*Process) } +/*The ComponentWithMustRun interface implements the MustRun method.*/ type ComponentWithMustRun interface { Component MustRun() bool diff --git a/core/inarrayport.go b/core/inarrayport.go index 53c1eb5..5ce1035 100644 --- a/core/inarrayport.go +++ b/core/inarrayport.go @@ -8,25 +8,26 @@ import ( type InArrayPort struct { network *Network - portName string - fullName string + //portName string + //fullName string array []*InPort mtx sync.Mutex closed bool + //downStrProc *Process } -func (c *InArrayPort) IsDrained() bool { +func (c *InArrayPort) isDrained() bool { for _, v := range c.array { - if !v.IsDrained() { + if !v.isDrained() { return false } } return true } -func (c *InArrayPort) IsEmpty() bool { +func (c *InArrayPort) isEmpty() bool { for _, v := range c.array { - if !v.IsEmpty() { + if !v.isEmpty() { return false } } @@ -37,9 +38,9 @@ func (c *InArrayPort) receive(p *Process) *Packet { panic("receive from an array port") } -func (c *InArrayPort) IsClosed() bool { +func (c *InArrayPort) isClosed() bool { for _, v := range c.array { - if !v.IsClosed() { + if !v.isClosed() { return false } } @@ -55,7 +56,7 @@ func (c *InArrayPort) GetArrayItem(i int) *InPort { return c.array[i] } -func (c *InArrayPort) SetArrayItem(c2 *InPort, i int) { +func (c *InArrayPort) setArrayItem(c2 *InPort, i int) { if i >= len(c.array) { // add to .array to fit c2 increaseBy := make([]*InPort, i-len(c.array)+1) diff --git a/core/initializationconnection.go b/core/initializationconnection.go index 88af991..bfb6949 100644 --- a/core/initializationconnection.go +++ b/core/initializationconnection.go @@ -1,33 +1,38 @@ package core import ( - //"fmt" + "fmt" "sync" ) type InitializationConnection struct { - network *Network - portName string - fullName string - closed bool - value interface{} - mtx sync.Mutex + network *Network + portName string + fullName string + closed bool + value interface{} + mtx sync.Mutex + downStrProc *Process } -func (c *InitializationConnection) IsDrained() bool { +func (c *InitializationConnection) isDrained() bool { c.mtx.Lock() defer c.mtx.Unlock() return c.closed } -func (c *InitializationConnection) IsEmpty() bool { +func (c *InitializationConnection) isEmpty() bool { c.mtx.Lock() defer c.mtx.Unlock() return !c.closed } +func (c *InitializationConnection) Receive(p *Process) *Packet { + return c.receive(c.downStrProc) +} + func (c *InitializationConnection) receive(p *Process) *Packet { if c.closed { @@ -39,7 +44,10 @@ func (c *InitializationConnection) receive(p *Process) *Packet { pkt.owner = p p.ownedPkts++ c.Close() - trace(p, " Received IIP: ", pkt.Contents.(string)) + trace(p, " Received IIP: ") + if tracing { + fmt.Print(" ", pkt.Contents, "\n") + } return pkt } @@ -51,7 +59,7 @@ func (c *InitializationConnection) Close() { } -func (c *InitializationConnection) IsClosed() bool { +func (c *InitializationConnection) isClosed() bool { c.mtx.Lock() defer c.mtx.Unlock() diff --git a/core/inport.go b/core/inport.go index 7a755a5..d246f35 100644 --- a/core/inport.go +++ b/core/inport.go @@ -1,7 +1,7 @@ package core import ( - //"fmt" + "fmt" "sync" "sync/atomic" ) @@ -20,12 +20,16 @@ type InPort struct { downStrProc *Process } +func (c *InPort) Receive(p *Process) *Packet { + return c.receive(c.downStrProc) +} + func (c *InPort) receive(p *Process) *Packet { LockTr(c.condNE, "recv L", p) defer UnlockTr(c.condNE, "recv U", p) trace(p, " Receiving from "+c.portName) - for c.isEmpty() { // InPort is empty - if c.closed { + for c.isEmptyNL() { // InPort is empty + if c.closed /* || c.upStrmCnt == 0 - check this! */ { trace(p, " Received end of stream from "+c.portName) return nil } @@ -35,11 +39,17 @@ func (c *InPort) receive(p *Process) *Packet { } pkt := c.pktArray[c.ir] c.pktArray[c.ir] = nil - if pkt.PktType != Normal { - trace(p, " Received from "+c.portName+" <", pkt.Contents.(string), - [...]string{"", "Open", "Close"}[pkt.PktType]) + if pkt.PktType != NormalPacket { + trace(p, " Received from "+c.portName+" < "+ + [...]string{"", "Open", "Close"}[pkt.PktType]+" Bracket") + if tracing { + fmt.Print(" contents: ", pkt.Contents, "\n") + } } else { - trace(p, " Received from "+c.portName+" <", pkt.Contents.(string)) + trace(p, " Received from "+c.portName+" < ") + if tracing { + fmt.Print(" ", pkt.Contents, "\n") + } } c.ir = (c.ir + 1) % len(c.pktArray) pkt.owner = p @@ -73,29 +83,29 @@ func (c *InPort) Close() { //c.downStrProc.activate() } -func (c *InPort) IsDrained() bool { +func (c *InPort) isDrained() bool { LockTr(c.condNE, "IDr L", c.downStrProc) defer UnlockTr(c.condNE, "IDr U", c.downStrProc) - return c.isDrained() + return c.isDrainedNL() } -func (c *InPort) isDrained() bool { - return c.isEmpty() && c.closed +func (c *InPort) isDrainedNL() bool { + return c.isEmptyNL() && c.closed } -func (c *InPort) IsEmpty() bool { +func (c *InPort) isEmpty() bool { LockTr(c.condNE, "IE L", c.downStrProc) defer UnlockTr(c.condNE, "IE U", c.downStrProc) - return c.isEmpty() + return c.isEmptyNL() } -func (c *InPort) isEmpty() bool { +func (c *InPort) isEmptyNL() bool { return c.ir == c.is && c.pktArray[c.is] == nil } -func (c *InPort) IsClosed() bool { +func (c *InPort) isClosed() bool { LockTr(c.condNE, "IC L", c.downStrProc) defer UnlockTr(c.condNE, "IC U", c.downStrProc) @@ -113,7 +123,7 @@ func (c *InPort) resetForNextExecution() { c.closed = false } -func (c *InPort) PktCount() int { +func (c *InPort) pktCount() int { var i int c.mtx.Lock() defer c.mtx.Unlock() diff --git a/core/inputconn.go b/core/inputconn.go index 3a5ecb8..edfb64a 100644 --- a/core/inputconn.go +++ b/core/inputconn.go @@ -1,10 +1,10 @@ package core type inputCommon interface { - IsDrained() bool + isDrained() bool resetForNextExecution() - IsEmpty() bool - IsClosed() bool + isEmpty() bool + isClosed() bool Close() receive(*Process) *Packet //Name() string @@ -20,6 +20,6 @@ type InputArrayConn interface { inputCommon GetArrayItem(i int) *InPort - SetArrayItem(c *InPort, i int) + setArrayItem(c *InPort, i int) ArrayLength() int } diff --git a/core/network.go b/core/network.go index 842dd3c..2d8eb69 100644 --- a/core/network.go +++ b/core/network.go @@ -1,23 +1,20 @@ package core import ( + "encoding/xml" + "errors" "fmt" - "io" - "os" + "io/ioutil" "regexp" "strconv" - "strings" "sync" "sync/atomic" ) -//var pStack []*Process - -//var stkLevel int - var tracing bool var tracelocks bool -var generate_gids bool +var tracepkts bool +var generateGids bool type Network struct { Name string @@ -25,21 +22,17 @@ type Network struct { //conns map[string]inputCommon wg sync.WaitGroup mother *Process + //params *Params } func NewNetwork(name string) *Network { net := &Network{ Name: name, procs: make(map[string]*Process), - //conns: make(map[string]inputCommon), - wg: sync.WaitGroup{}, + wg: sync.WaitGroup{}, + //params: pms, } - //stkLevel++ - //if stkLevel >= len(pStack) { - //pStack = append(pStack, nil) - //} - return net } @@ -47,8 +40,7 @@ func NewSubnet(Name string, p *Process) *Network { net := &Network{ Name: Name, procs: make(map[string]*Process), - //conns: make(map[string]inputCommon), - wg: sync.WaitGroup{}, + wg: sync.WaitGroup{}, } net.mother = p @@ -91,20 +83,18 @@ func WaitTr(sc *sync.Cond, s string, p *Process) { sc.Wait() } -func trace(p *Process, s ...string) { +func trace(p *Process, s string) { if tracing { - fmt.Print(p.Name, " "+strings.Trim(fmt.Sprint(s), "[]")+"\n") + fmt.Print(p.Name, " "+s+"\n") } } -func traceNet(n *Network, s ...string) { +func traceNet(n *Network, s string) { if tracing { - fmt.Print(n.Name, " "+strings.Trim(fmt.Sprint(s), "[]")+"\n") + fmt.Print(n.Name, " "+s+"\n") } } -//func (n *Network) id() string { return fmt.Sprintf("%p", n) } - func (n *Network) NewProc(nm string, comp Component) *Process { proc := &Process{ @@ -189,7 +179,7 @@ func (n *Network) Connect(p1 *Process, out string, p2 *Process, in string, cap i if anyInConn == nil { p2.inPorts[inPort.name] = connxn } else { - anyInConn.(InputArrayConn).SetArrayItem(connxn, inPort.index) + anyInConn.(InputArrayConn).setArrayItem(connxn, inPort.index) } } } else { @@ -228,7 +218,7 @@ func (n *Network) Connect(p1 *Process, out string, p2 *Process, in string, cap i //p1.outPorts[out] = anyOutConn //opt.Name = out opt = new(OutPort) - outConn.SetArrayItem(opt, outPort.index) + outConn.setArrayItem(opt, outPort.index) opt.portName = out opt.fullName = p1.Name + "." + out //n.conns[opt.fullName] = opt.Conn @@ -244,7 +234,7 @@ func (n *Network) Connect(p1 *Process, out string, p2 *Process, in string, cap i } opt.sender = p1 - opt.Conn = connxn + opt.conn = connxn opt.connected = true connxn.incUpstream() @@ -295,52 +285,44 @@ func (n *Network) Exit() { } } -func setOptions() { - var rec string - f, err := os.Open("params.xml") - if err == nil { - defer f.Close() - buf := make([]byte, 1024) - for { - n, err := f.Read(buf) - if err == io.EOF { - break - } - if err != nil { - fmt.Println(err) - continue - } - if n > 0 { - //fmt.Println(string(buf[:n])) - rec += string(buf[:n]) - } - } +type Params struct { + Tracing bool `xml:"tracing"` + TraceLocks bool `xml:"tracelocks"` + TracePkts bool `xml:"tracepkts"` + GenerateGIDs bool `xml:"generate-gIds"` +} - s := "" - i := strings.Index(rec, s) - if i > -1 && rec[i+len(s):i+len(s)+4] == "true" { - tracing = true - } +func LoadXMLParams(s string) (*Params, error) { - s = "" - i = strings.Index(rec, s) - if i > -1 && rec[i+len(s):i+len(s)+4] == "true" { - tracelocks = true - } + params := &Params{} - s = "" - i = strings.Index(rec, s) - if i > -1 && rec[i+len(s):i+len(s)+4] == "true" { - generate_gids = true - } + xmldata, err := ioutil.ReadFile(s) + if err != nil { + return nil, errors.New("couldn't read: " + s) + } + err = xml.Unmarshal(xmldata, ¶ms) + if err != nil { + return nil, errors.New("couldn't unmarshal: " + s) + } + + return params, nil +} + +func (n *Network) SetParams(p *Params) { + if p == nil { + panic("Calling SetParams with nil parameter") } + tracing = p.Tracing + tracelocks = p.TraceLocks + tracepkts = p.TracePkts + generateGids = p.GenerateGIDs } func (n *Network) Run() { defer n.Exit() - if n.mother == nil { - setOptions() - } + //if n.mother == nil { + // n.setOptions() + //} defer traceNet(n, " Done") @@ -410,12 +392,7 @@ func (n *Network) Run() { } } - var p *Process - //if n.mother != nil { - p = n.mother - //} else { - // return - //} + p := n.mother allDrained, _, _ := p.inputState() if allDrained { @@ -428,12 +405,12 @@ func (n *Network) Run() { _, b := v.(*OutArrayPort) if b { for _, w := range v.(*OutArrayPort).array { - w.Conn.upStrmCnt = 0 + w.conn.upStrmCnt = 0 } } else { w, b := v.(*OutPort) if b { - w.Conn.upStrmCnt = 0 + w.conn.upStrmCnt = 0 } } } @@ -444,12 +421,12 @@ func (n *Network) Run() { _, b := v.(*OutArrayPort) if b { for _, w := range v.(*OutArrayPort).array { - w.Conn.incUpstream() + w.conn.incUpstream() } } else { w, b := v.(*OutPort) if b { - w.Conn.incUpstream() + w.conn.incUpstream() } } } diff --git a/core/nulloutport.go b/core/nulloutport.go index 8903f28..d975644 100644 --- a/core/nulloutport.go +++ b/core/nulloutport.go @@ -1,9 +1,9 @@ package core type NullOutPort struct { - sender *Process - portName string - fullName string + //sender *Process + //portName string + //fullName string } // NullOutPort by default discards the packet. diff --git a/core/outarrayport.go b/core/outarrayport.go index 119bf41..a9cb1e0 100644 --- a/core/outarrayport.go +++ b/core/outarrayport.go @@ -25,7 +25,7 @@ func (o *OutArrayPort) GetArrayItem(i int) *OutPort { return o.array[i] } -func (o *OutArrayPort) SetArrayItem(o2 *OutPort, i int) { +func (o *OutArrayPort) setArrayItem(o2 *OutPort, i int) { if i >= len(o.array) { // add to .array to fit c2 increaseBy := make([]*OutPort, i-len(o.array)+1) @@ -53,7 +53,7 @@ func (o *OutArrayPort) GetItemWithFewestIPs() int { backlog := math.MaxInt32 var fewestIPsIndex int for i, v := range o.array { - j := v.Conn.PktCount() + j := v.conn.pktCount() //fmt.Println(i, j) if j <= backlog { fewestIPsIndex = i diff --git a/core/outport.go b/core/outport.go index eb056a0..6be612d 100644 --- a/core/outport.go +++ b/core/outport.go @@ -1,13 +1,14 @@ package core import ( + "fmt" "sync/atomic" ) type OutPort struct { portName string fullName string - Conn *InPort + conn *InPort connected bool sender *Process network *Network @@ -25,35 +26,39 @@ func (o *OutPort) send(p *Process, pkt *Packet) bool { panic("Sending packet not owned by this process") } - LockTr(o.Conn.condNF, "send L", p) - defer UnlockTr(o.Conn.condNF, "send U", p) + LockTr(o.conn.condNF, "send L", p) + defer UnlockTr(o.conn.condNF, "send U", p) - if pkt.PktType != Normal { - trace(p, " Sending to "+o.portName+" >", pkt.Contents.(string), - [...]string{"", "Open", "Close"}[pkt.PktType]) + if pkt.PktType != NormalPacket { + trace(p, " Sending to "+o.portName+" > "+ + [...]string{"", "Open", "Close"}[pkt.PktType]+" Bracket") + if tracing { + fmt.Print(" contents: ", pkt.Contents, "\n") + } } else { - trace(p, " Sending to "+o.portName+" >", pkt.Contents.(string)) + trace(p, " Sending to "+o.portName+" > ") + if tracing { + fmt.Print(" ", pkt.Contents, "\n") + } } - for o.Conn.isFull() { // while connection is full + for o.conn.isFull() { // while connection is full atomic.StoreInt32(&p.status, SuspSend) - WaitTr(o.Conn.condNF, "wait in send", p) + WaitTr(o.conn.condNF, "wait in send", p) atomic.StoreInt32(&p.status, Active) } trace(p, " Sent to "+o.portName) - trace(o.Conn.downStrProc, "act from send") - //LockTr(o.Conn.downStrProc.canGo, "start test L", o.Conn.downStrProc) - o.Conn.downStrProc.activate() - //UnlockTr(o.Conn.downStrProc.canGo, "start test U", o.Conn.downStrProc) + trace(o.conn.downStrProc, "act from send") + o.conn.downStrProc.activate() - o.Conn.pktArray[o.Conn.is] = pkt - o.Conn.is = (o.Conn.is + 1) % len(o.Conn.pktArray) + o.conn.pktArray[o.conn.is] = pkt + o.conn.is = (o.conn.is + 1) % len(o.conn.pktArray) //pkt.owner = nil p.ownedPkts-- pkt = nil - BdcastTr(o.Conn.condNE, "bdcast sent", p) + BdcastTr(o.conn.condNE, "bdcast sent", p) return true } @@ -75,17 +80,15 @@ func (o *OutPort) ArrayLength() int { } func (o *OutPort) Close() { - LockTr(o.Conn.condNF, "close L", o.sender) - defer UnlockTr(o.Conn.condNF, "close U", o.sender) + LockTr(o.conn.condNF, "close L", o.sender) + defer UnlockTr(o.conn.condNF, "close U", o.sender) trace(o.sender, " Close "+o.portName) - o.Conn.decUpstream() - if o.Conn.upStrmCnt == 0 { - o.Conn.closed = true - BdcastTr(o.Conn.condNE, "bdcast out", o.sender) - trace(o.Conn.downStrProc, "act from close") - //LockTr(o.Conn.downStrProc.canGo, "start test L", o.Conn.downStrProc) - o.Conn.downStrProc.activate() - //UnlockTr(o.Conn.downStrProc.canGo, "start test U", o.Conn.downStrProc) + o.conn.decUpstream() + if o.conn.upStrmCnt == 0 { + o.conn.closed = true + BdcastTr(o.conn.condNE, "bdcast out", o.sender) + trace(o.conn.downStrProc, "activated from close") + o.conn.downStrProc.activate() } } diff --git a/core/outputconn.go b/core/outputconn.go index a4e5a7b..3ad50d0 100644 --- a/core/outputconn.go +++ b/core/outputconn.go @@ -14,7 +14,7 @@ type OutputConn interface { type OutputArrayConn interface { outputCommon GetArrayItem(i int) *OutPort - SetArrayItem(c *OutPort, i int) + setArrayItem(c *OutPort, i int) ArrayLength() int GetItemWithFewestIPs() int } diff --git a/core/packet.go b/core/packet.go index f5cd950..0906ae8 100644 --- a/core/packet.go +++ b/core/packet.go @@ -1,7 +1,7 @@ package core const ( - Normal int32 = iota + NormalPacket int32 = iota OpenBracket CloseBracket ) diff --git a/core/process.go b/core/process.go index 652cda5..8ce0af3 100644 --- a/core/process.go +++ b/core/process.go @@ -9,6 +9,7 @@ import ( "sync/atomic" ) +//Process type defines a gofbp process. type Process struct { gid uint64 Name string @@ -27,6 +28,7 @@ type Process struct { autoOutput inputCommon } +//Nostarted const reflects current process status const ( Notstarted int32 = iota Active @@ -36,6 +38,7 @@ const ( Terminated ) +//OpenInPort function opens and returns InputConn func (p *Process) OpenInPort(s string) InputConn { var in InputConn var b bool @@ -50,6 +53,7 @@ func (p *Process) OpenInPort(s string) InputConn { return in } +//OpenInArrayPort method opens and returns InArrayPort func (p *Process) OpenInArrayPort(s string) *InArrayPort { var in *InArrayPort var b bool @@ -66,6 +70,7 @@ func (p *Process) OpenInArrayPort(s string) *InArrayPort { return in } +//OpenOutPort method opens and returns OutputConn func (p *Process) OpenOutPort(s string) OutputConn { var out OutputConn var b bool @@ -85,6 +90,7 @@ func (p *Process) OpenOutPort(s string) OutputConn { } +//OpenOutPortOptional function opens and returns OutputConn func (p *Process) OpenOutPortOptional(s string) OutputConn { var out OutputConn var b bool @@ -108,6 +114,7 @@ func (p *Process) OpenOutPortOptional(s string) OutputConn { // not sure it makes sense to allow optional for array ports! +//OpenOutArrayPort method opens and returns OutArrayPort func (p *Process) OpenOutArrayPort(s string) *OutArrayPort { var out *OutArrayPort var b bool @@ -128,18 +135,21 @@ func (p *Process) OpenOutArrayPort(s string) *OutArrayPort { } // Send sends a packet to the output connection. -// Returns false when fails to send. + +//Send method emits Packet, returning false when fails to send. func (p *Process) Send(o OutputConn, pkt *Packet) bool { //o.SetSender(p) return o.send(p, pkt) } +//Receive method accepts InputConn and returns Packet // Receive receives from the connection. // Returns nil, when there's no more data. func (p *Process) Receive(c InputConn) *Packet { return c.receive(p) } +//Close method closes InputConn func (p *Process) Close(c InputConn) { c.Close() } @@ -184,15 +194,15 @@ func (p *Process) inputState() (bool, bool, bool) { _, b := v.(*InArrayPort) if b { for _, w := range v.(*InArrayPort).array { - allDrained = allDrained && w.IsDrained() - hasData = hasData || !w.IsEmpty() + allDrained = allDrained && w.isDrained() + hasData = hasData || !w.isEmpty() selfStarting = false } } else { w, b := v.(*InPort) if b { - allDrained = allDrained && v.IsDrained() - hasData = hasData || !w.IsEmpty() + allDrained = allDrained && v.isDrained() + hasData = hasData || !w.isEmpty() selfStarting = false } } @@ -208,13 +218,14 @@ func (p *Process) inputState() (bool, bool, bool) { } } +//Run method initializes and Executes Process func (p *Process) Run() { defer atomic.StoreInt32(&p.status, Terminated) defer trace(p, " terminated") trace(p, " started") - if generate_gids { + if generateGids { fmt.Println("Goroutine", p.Name+":", "no.", getGID()) } @@ -279,15 +290,21 @@ func (p *Process) isMustRun() bool { return hasMustRun } +//Create method creates and interface and returns a Packet // create packet containing anything! func (p *Process) Create(x interface{}) *Packet { var pkt *Packet = new(Packet) pkt.Contents = x pkt.owner = p p.ownedPkts++ + if tracepkts { + fmt.Print(p.Name, " Packet created <\n") + fmt.Println(" ", pkt.Contents) + } return pkt } +//CreateBracket method builds a new Packet and returns it // create bracket func (p *Process) CreateBracket(pktType int32, s string) *Packet { var pkt *Packet = new(Packet) @@ -295,14 +312,28 @@ func (p *Process) CreateBracket(pktType int32, s string) *Packet { pkt.PktType = pktType pkt.owner = p p.ownedPkts++ + if tracepkts { + fmt.Print(p.Name, " Bracket created: ", [...]string{"", "Open", "Close"}[pkt.PktType]+" Bracket") + fmt.Println(" ", pkt.Contents) + } return pkt } +//Discard method sets Packet to nil func (p *Process) Discard(pkt *Packet) { if pkt == nil { panic("Discarding nil packet") } p.ownedPkts-- + if tracepkts { + if pkt.PktType != NormalPacket { + fmt.Print(p.Name, " Bracket discarded: ", [...]string{"", "Open", "Close"}[pkt.PktType]+" Bracket") + fmt.Print(" contents: ", pkt.Contents, "\n") + } else { + fmt.Print(p.Name, " Packet discarded >") + fmt.Println(" ", pkt.Contents) + } + } pkt = nil } diff --git a/core/subnet_support.go b/core/subnet_support.go index de87af5..6cc982c 100644 --- a/core/subnet_support.go +++ b/core/subnet_support.go @@ -7,7 +7,7 @@ import ( type SubIn struct { //ipt InputConn - iptIp InputConn + iptIP InputConn out OutputConn eipt InputConn } @@ -22,12 +22,12 @@ func (subIn *SubIn) Setup(p *Process) { subIn.out = p.OpenOutPort("OUT") - subIn.iptIp = p.OpenInPort("NAME") + subIn.iptIP = p.OpenInPort("NAME") } func (subIn *SubIn) Execute(p *Process) { - icpkt := p.Receive(subIn.iptIp) + icpkt := p.Receive(subIn.iptIP) param := icpkt.Contents.(string) p.Discard(icpkt) @@ -56,7 +56,7 @@ func (subIn *SubIn) Execute(p *Process) { func (subIn *SubInSS) Execute(p *Process) { - icpkt := p.Receive(subIn.iptIp) + icpkt := p.Receive(subIn.iptIP) param := icpkt.Contents.(string) p.Discard(icpkt) @@ -92,7 +92,7 @@ func (subIn *SubInSS) Execute(p *Process) { type SubOut struct { ipt InputConn - iptIp InputConn + iptIP InputConn eopt OutputConn } @@ -104,14 +104,14 @@ func (subOut *SubOut) Setup(p *Process) { subOut.ipt = p.OpenInPort("IN") - subOut.iptIp = p.OpenInPort("NAME") + subOut.iptIP = p.OpenInPort("NAME") } //func (SubOut) MustRun() {} func (subOut *SubOut) Execute(p *Process) { - icpkt := p.Receive(subOut.iptIp) + icpkt := p.Receive(subOut.iptIP) param := icpkt.Contents.(string) p.Discard(icpkt) @@ -137,7 +137,7 @@ func (subOut *SubOut) Execute(p *Process) { } func (subOut *SubOutSS) Execute(p *Process) { - icpkt := p.Receive(subOut.iptIp) + icpkt := p.Receive(subOut.iptIP) param := icpkt.Contents.(string) p.Discard(icpkt) diff --git a/diagrams/Esses.png b/diagrams/Esses.png new file mode 100644 index 0000000..cc4772d Binary files /dev/null and b/diagrams/Esses.png differ diff --git a/diagrams/Merge.drw b/diagrams/Merge.drw new file mode 100644 index 0000000..329fcf8 --- /dev/null +++ b/diagrams/Merge.drw @@ -0,0 +1,26 @@ + + + + + 277 411 2 B 92 64 Sender2 sender.go C:/Users/Paul/Documents/GitHub/gofbp/components/testrtn/sender.go falsefalsefalse + + 555 316 3 B 92 64 Write + To +Console writetoconsole.go C:/Users/Paul/Documents/GitHub/gofbp/components/testrtn/writetoconsole.go falsefalsefalse + + 277 235 4 B 92 64 Sender1 sender.go C:/Users/Paul/Documents/GitHub/gofbp/components/testrtn/sender.go falsefalsefalse + + 100 234 5 I 52 30 15 falsefalsefalse + + 102 406 6 I 52 30 10 falsefalsefalse + + + 323 232 509 312 4 3 4 falseOUTIN0 424 232 + 424 312 + + 126 234 231 233 5 4 7 falseCOUNT0 + 128 403 231 402 6 2 8 falseCOUNT0 + 323 409 431 314 2 4 12 trueOUT2 430 409 + + \ No newline at end of file diff --git a/diagrams/Subnet-diagram.drw b/diagrams/Subnet-diagram.drw index 8752022..aab843b 100644 --- a/diagrams/Subnet-diagram.drw +++ b/diagrams/Subnet-diagram.drw @@ -4,28 +4,19 @@ xsi:noNamespaceSchemaLocation="https://github.com/jpaulm/drawfbp/blob/master/lib 116 192 1 B 92 64 Substream - Sender falsefalsefalse + Sender sender.go C:/Users/Paul/Documents/GitHub/gofbp/components/testrtn/sender.go falsefalsefalse 375 192 2 B 92 64 Run - Subnet falsefalsefalse + Subnet C:/Users/Paul/Documents/GitHub/gofbp/diagrams/Subnet1.drw subnet1.go C:/Users/Paul/Documents/GitHub/gofbp/subnet_tests/subnets/subnet1.go falsefalsetrue 645 192 3 B 92 64 Write To -Console falsefalsefalse +Console writetoconsole.go C:/Users/Paul/Documents/GitHub/gofbp/components/testrtn/writetoconsole.go falsefalsefalse - 179 368 4 B 92 64 SubInSS falsefalsefalse - - 375 368 5 B 92 64 Write To -Console falsefalsefalse - - 578 368 6 B 92 64 SubOutSS falsefalsefalse - - 119 64 8 I 68 30 40 falsefalsefalse + 119 64 8 I 68 30 40 falsefalsefalse 162 190 329 190 1 2 1 falseOUTIN0 421 191 599 191 2 3 2 falseOUTIN0 - 225 365 329 365 4 5 3 falseOUTIN0 - 421 366 532 366 5 6 4 falseOUTIN0 - 114 79 114 160 8 1 6 falseIN0 + 114 79 114 160 8 1 6 falseCOUNT0 \ No newline at end of file diff --git a/diagrams/Subnet-diagram.png b/diagrams/Subnet-diagram.png index 2b71154..d9d475e 100644 Binary files a/diagrams/Subnet-diagram.png and b/diagrams/Subnet-diagram.png differ diff --git a/diagrams/Subnet1.drw b/diagrams/Subnet1.drw new file mode 100644 index 0000000..e7097e5 --- /dev/null +++ b/diagrams/Subnet1.drw @@ -0,0 +1,17 @@ + + + + + 365 160 9 B 92 64 Write + To +Console writetoconsnl.go C:/Users/Paul/Documents/GitHub/gofbp/components/testrtn/writetoconsnl.go falsefalsefalse + + 111 154 10 C 36 28 IN truefalsefalsefalse + + 632 156 13 D 36 28 OUT truefalsefalsefalse + + + 129 156 319 156 10 9 7 falseOUTIN0 + 411 159 614 158 9 13 11 falseOUTIN0 + \ No newline at end of file diff --git a/go.mod b/go.mod index dfc619e..e33c830 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,6 @@ module github.com/jpaulm/gofbp go 1.17 + +require github.com/gorilla/websocket v1.4.2 // indirect + diff --git a/go.sum b/go.sum index db93090..85efffd 100644 --- a/go.sum +++ b/go.sum @@ -1,2 +1,2 @@ -github.com/jpaulm/gofbp v1.0.0 h1:ly0gtz/ihVuu1H71Z4iRLmz+n9AfqEkJHgTkIkxPEtc= -github.com/jpaulm/gofbp v1.0.0/go.mod h1:g2Dwwa5o9SUdYZCix8jfHG6N4dl1p2WNGYBsYfs6uZU= +github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc= +github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= diff --git a/inarrayport.go b/inarrayport.go new file mode 100644 index 0000000..3f39f94 --- /dev/null +++ b/inarrayport.go @@ -0,0 +1,79 @@ +package gofbp + +import ( + //"fmt" + "sync" +) + +type InArrayPort struct { + network *Network + + portName string + fullName string + array []*InPort + mtx sync.Mutex + closed bool +} + +func (c *InArrayPort) IsDrained() bool { + for _, v := range c.array { + if !v.IsDrained() { + return false + } + } + return true +} + +func (c *InArrayPort) IsEmpty() bool { + for _, v := range c.array { + if !v.IsEmpty() { + return false + } + } + return true +} + +func (c *InArrayPort) receive(p *Process) *Packet { + panic("receive from an array port") +} + +func (c *InArrayPort) IsClosed() bool { + for _, v := range c.array { + if !v.IsClosed() { + return false + } + } + return true +} + +func (c *InArrayPort) resetForNextExecution() {} + +func (c *InArrayPort) GetArrayItem(i int) *InPort { + if i >= len(c.array) { + return nil + } + return c.array[i] +} + +func (c *InArrayPort) SetArrayItem(c2 *InPort, i int) { + if i >= len(c.array) { + // add to .array to fit c2 + increaseBy := make([]*InPort, i-len(c.array)+1) + c.array = append(c.array, increaseBy...) + } + c.array[i] = c2 +} + +func (c *InArrayPort) ArrayLength() int { + return len(c.array) +} + +func (c *InArrayPort) Close() { + c.mtx.Lock() + defer c.mtx.Unlock() + c.closed = true + for _, w := range c.array { + w.closed = true + BdcastTr(w.condNE, "bdcast in array NE", w.downStrProc) + } +} diff --git a/initializationconnection.go b/initializationconnection.go new file mode 100644 index 0000000..e4e3419 --- /dev/null +++ b/initializationconnection.go @@ -0,0 +1,66 @@ +package gofbp + +import ( + //"fmt" + "sync" +) + +type InitializationConnection struct { + network *Network + portName string + fullName string + closed bool + value interface{} + mtx sync.Mutex +} + +func (c *InitializationConnection) IsDrained() bool { + c.mtx.Lock() + defer c.mtx.Unlock() + + return c.closed +} + +func (c *InitializationConnection) IsEmpty() bool { + c.mtx.Lock() + defer c.mtx.Unlock() + + return !c.closed +} + +func (c *InitializationConnection) receive(p *Process) *Packet { + + if c.closed { + return nil + } + trace(p, " Receiving IIP") + var pkt *Packet = new(Packet) + pkt.Contents = c.value + pkt.owner = p + p.ownedPkts++ + c.Close() + trace(p, " Received IIP: ", pkt.Contents.(string)) + return pkt +} + +func (c *InitializationConnection) Close() { + c.mtx.Lock() + defer c.mtx.Unlock() + + c.closed = true + +} + +func (c *InitializationConnection) IsClosed() bool { + c.mtx.Lock() + defer c.mtx.Unlock() + + return c.closed +} + +func (c *InitializationConnection) resetForNextExecution() { + c.mtx.Lock() + defer c.mtx.Unlock() + + c.closed = false +} diff --git a/inport.go b/inport.go new file mode 100644 index 0000000..6299b09 --- /dev/null +++ b/inport.go @@ -0,0 +1,126 @@ +package gofbp + +import ( + //"fmt" + "sync" + "sync/atomic" +) + +type InPort struct { + network *Network + pktArray []*Packet + is, ir int // send index and receive index + mtx sync.Mutex + condNE *sync.Cond + condNF *sync.Cond + closed bool + upStrmCnt int + portName string + fullName string + downStrProc *Process +} + +func (c *InPort) receive(p *Process) *Packet { + LockTr(c.condNE, "recv L", p) + defer UnlockTr(c.condNE, "recv U", p) + trace(p, " Receiving from "+c.portName) + for c.isEmpty() { // InPort is empty + if c.closed { + trace(p, " Received end of stream from "+c.portName) + return nil + } + atomic.StoreInt32(&p.status, SuspRecv) + WaitTr(c.condNE, "wait in recv", p) + atomic.StoreInt32(&p.status, Active) + } + pkt := c.pktArray[c.ir] + c.pktArray[c.ir] = nil + if pkt.PktType != Normal { + trace(p, " Received from "+c.portName+" <", pkt.Contents.(string), + [...]string{"", "Open", "Close"}[pkt.PktType]) + } else { + trace(p, " Received from "+c.portName+" <", pkt.Contents.(string)) + } + c.ir = (c.ir + 1) % len(c.pktArray) + pkt.owner = p + p.ownedPkts++ + //c.condNF.Broadcast() + BdcastTr(c.condNF, "bdcast recv'd", p) + + return pkt +} + +func (c *InPort) incUpstream() { + LockTr(c.condNE, "IUS L", nil) + defer UnlockTr(c.condNE, "IUS U", nil) + + c.upStrmCnt++ +} + +func (c *InPort) decUpstream() { + //LockTr(c.condNE, "DUS L", nil) + //defer UnlockTr(c.condNE, "DUS U", nil) + c.upStrmCnt-- +} + +func (c *InPort) Close() { + LockTr(c.condNE, "ClsI L", c.downStrProc) + defer UnlockTr(c.condNE, "ClsI U", c.downStrProc) + + c.closed = true + //c.condNE.Broadcast() + BdcastTr(c.condNE, "bdcast in NE", c.downStrProc) + //c.downStrProc.activate() +} + +func (c *InPort) IsDrained() bool { + LockTr(c.condNE, "IDr L", c.downStrProc) + defer UnlockTr(c.condNE, "IDr U", c.downStrProc) + + return c.isDrained() +} + +func (c *InPort) isDrained() bool { + return c.isEmpty() && c.closed +} + +func (c *InPort) IsEmpty() bool { + LockTr(c.condNE, "IE L", c.downStrProc) + defer UnlockTr(c.condNE, "IE U", c.downStrProc) + + return c.isEmpty() +} + +func (c *InPort) isEmpty() bool { + return c.ir == c.is && c.pktArray[c.is] == nil +} + +func (c *InPort) IsClosed() bool { + LockTr(c.condNE, "IC L", c.downStrProc) + defer UnlockTr(c.condNE, "IC U", c.downStrProc) + + return c.closed +} + +func (c *InPort) isFull() bool { + return c.ir == c.is && c.pktArray[c.is] != nil +} + +func (c *InPort) resetForNextExecution() { + c.mtx.Lock() + defer c.mtx.Unlock() + + c.closed = false +} + +func (c *InPort) PktCount() int { + var i int + c.mtx.Lock() + defer c.mtx.Unlock() + for _, p := range c.pktArray { + if p != nil { + i++ + } + } + return i +} diff --git a/inputconn.go b/inputconn.go new file mode 100644 index 0000000..2f07f1c --- /dev/null +++ b/inputconn.go @@ -0,0 +1,25 @@ +package gofbp + +type inputCommon interface { + IsDrained() bool + resetForNextExecution() + IsEmpty() bool + IsClosed() bool + Close() + receive(*Process) *Packet + //Name() string + //SetName(string) +} + +type InputConn interface { + inputCommon + //receive(p *Process) *Packet +} + +type InputArrayConn interface { + inputCommon + + GetArrayItem(i int) *InPort + SetArrayItem(c *InPort, i int) + ArrayLength() int +} diff --git a/internal/factor.sh b/internal/factor.sh new file mode 100755 index 0000000..208268d --- /dev/null +++ b/internal/factor.sh @@ -0,0 +1,21 @@ +#!/bin/bash + +pfx="github.com/" +from="$pfx$1"; to="$pfx$2" +log=/tmp/factor.log + +case $# in + 2) + echo FACTOR.SH: `date` >> $log + for sdir in ../ ../io ../subnets ../testrtn ../utils ../internal ; do + pushd $sdir >/dev/null + for fl in *go; do + echo $sdir/$fl >> $log + sed -s s"!$from!$to!" $fl > /tmp/$fl && mv /tmp/$fl $fl && count=$(($count+1)) + done + popd >/dev/null + done + echo $count hits + ;; + *)echo USAGE: $0 FROM TO; exit 1;; +esac diff --git a/force_deadlock_test.go b/internal/force_deadlock_test.go similarity index 100% rename from force_deadlock_test.go rename to internal/force_deadlock_test.go diff --git a/internal/gofbp_test.go b/internal/gofbp_test.go new file mode 100644 index 0000000..f507e46 --- /dev/null +++ b/internal/gofbp_test.go @@ -0,0 +1,139 @@ +package main + +import ( + "testing" + + "path/filepath" + + "github.com/jpaulm/gofbp" + "github.com/jpaulm/gofbp/io" + "github.com/jpaulm/gofbp/testrtn" +) + +func TestMerge(t *testing.T) { + net := gofbp.NewNetwork("Merge") + + proc1 := net.NewProc("Sender1", &testrtn.Sender{}) + proc2 := net.NewProc("Sender2", &testrtn.Sender{}) + + proc3 := net.NewProc("WriteToConsole", &testrtn.WriteToConsole{}) + + net.Initialize("15", proc1, "COUNT") + net.Initialize("10", proc2, "COUNT") + net.Connect(proc1, "OUT", proc3, "IN", 6) + net.Connect(proc2, "OUT", proc3, "IN", 6) + + net.Run() +} + +func TestConcat(t *testing.T) { + net := gofbp.NewNetwork("Concat") + + proc1 := net.NewProc("Sender", &testrtn.Sender{}) + + proc1a := net.NewProc("Sender2", &testrtn.Sender{}) + + proc2 := net.NewProc("ConcatStr", &testrtn.ConcatStr{}) + + proc3 := net.NewProc("WriteToConsole", &testrtn.WriteToConsole{}) + + net.Initialize("15", proc1, "COUNT") + net.Initialize("10", proc1a, "COUNT") + net.Connect(proc1, "OUT", proc2, "IN[0]", 6) + net.Connect(proc1a, "OUT", proc2, "IN[1]", 6) + net.Connect(proc2, "OUT", proc3, "IN", 6) + + net.Run() +} + +func TestRRDist(t *testing.T) { + net := gofbp.NewNetwork("RRDist") + + proc1 := net.NewProc("Sender", &testrtn.Sender{}) + + proc2 := net.NewProc("RoundRobinSender", &testrtn.RoundRobinSender{}) + + proc3a := net.NewProc("WriteToConsole", &testrtn.WriteToConsole{}) + proc3b := net.NewProc("Receiver1", &testrtn.Receiver{}) + proc3c := net.NewProc("Receiver2", &testrtn.Receiver{}) + + net.Initialize("15", proc1, "COUNT") + net.Connect(proc1, "OUT", proc2, "IN", 6) + net.Connect(proc2, "OUT[0]", proc3a, "IN", 6) + net.Connect(proc2, "OUT[1]", proc3b, "IN", 6) + net.Connect(proc2, "OUT[2]", proc3c, "IN", 6) + + net.Run() +} + +func TestCopyFile(t *testing.T) { + net := gofbp.NewNetwork("CopyFile") + + proc1 := net.NewProc("ReadFile", &io.ReadFile{}) + + proc2 := net.NewProc("WriteFile", &io.WriteFile{}) + + net.Initialize(filepath.Join("testdata", "testdata.txt"), proc1, "FILENAME") + net.Initialize(filepath.Join("testdata", "copy-file.tmp"), proc2, "FILENAME") + net.Connect(proc1, "OUT", proc2, "IN", 6) + + net.Run() +} + +func TestDoSelect1(t *testing.T) { + + // port REJ from proc2 is connected + + net := gofbp.NewNetwork("DoSelect1") + + proc1 := net.NewProc("ReadFile", &io.ReadFile{}) + proc2 := net.NewProc("Select", &testrtn.Selector{}) + proc3a := net.NewProc("WriteFile", &io.WriteFile{}) + proc3b := net.NewProc("WriteToConsole", &testrtn.WriteToConsole{}) + + net.Initialize(filepath.Join("testdata", "testdata.txt"), proc1, "FILENAME") + net.Initialize("X", proc2, "PARAM") + net.Initialize(filepath.Join("testdata", "do-select.tmp"), proc3a, "FILENAME") + net.Connect(proc1, "OUT", proc2, "IN", 6) + net.Connect(proc2, "ACC", proc3a, "IN", 6) + net.Connect(proc2, "REJ", proc3b, "IN", 6) + + net.Run() +} + +func TestDoSelect2(t *testing.T) { + + // port REJ from proc2 is NOT connected + + net := gofbp.NewNetwork("DoSelect2") + + proc1 := net.NewProc("ReadFile", &io.ReadFile{}) + proc2 := net.NewProc("Select", &testrtn.Selector{}) + proc3a := net.NewProc("WriteFile", &io.WriteFile{}) + //proc3b := net.NewProc("WriteToConsole", &testrtn.WriteToConsole{}) + + net.Initialize(filepath.Join("testdata", "testdata.txt"), proc1, "FILENAME") + net.Initialize("X", proc2, "PARAM") + net.Initialize(filepath.Join("testdata", "do-select.tmp"), proc3a, "FILENAME") + net.Connect(proc1, "OUT", proc2, "IN", 6) + net.Connect(proc2, "ACC", proc3a, "IN", 6) + //net.Connect(proc2, "REJ", proc3b, "IN", 6) + + net.Run() +} +func TestWriteToConsUsingNL(t *testing.T) { + + net := gofbp.NewNetwork("TestWriteToConsUsingNL") + + proc1 := net.NewProc("Sender1", &testrtn.Sender{}) + proc2 := net.NewProc("Sender2", &testrtn.Sender{}) + + proc3 := net.NewProc("WriteToConsNL", &testrtn.WriteToConsNL{}) + + net.Initialize("15", proc1, "COUNT") + net.Initialize("15", proc2, "COUNT") + net.Connect(proc1, "OUT", proc3, "IN", 6) + net.Connect(proc2, "OUT", proc3, "IN", 6) + + net.Run() +} diff --git a/internal/params.xml b/internal/params.xml new file mode 100644 index 0000000..2bf5e9b --- /dev/null +++ b/internal/params.xml @@ -0,0 +1,6 @@ + + +false +false +false + diff --git a/internal/renovate.json b/internal/renovate.json new file mode 100644 index 0000000..1eb0985 --- /dev/null +++ b/internal/renovate.json @@ -0,0 +1,9 @@ +{ + "extends": [ + "config:base", ":disableDependencyDashboard" + ] , + "automerge": true, + "major": { + "automerge": false + } +} diff --git a/internal/settings.json b/internal/settings.json new file mode 100644 index 0000000..fb15a9c --- /dev/null +++ b/internal/settings.json @@ -0,0 +1,3 @@ +{ +"go.testFlags": ["-race"] +} \ No newline at end of file diff --git a/testdata/deadlock/main.go b/internal/testdata/deadlock/main.go similarity index 100% rename from testdata/deadlock/main.go rename to internal/testdata/deadlock/main.go diff --git a/testdata/infqueue b/internal/testdata/infqueue similarity index 100% rename from testdata/infqueue rename to internal/testdata/infqueue diff --git a/testdata/testdata.txt b/internal/testdata/testdata.txt similarity index 100% rename from testdata/testdata.txt rename to internal/testdata/testdata.txt diff --git a/internal/testinfqueueasmain_test.go b/internal/testinfqueueasmain_test.go new file mode 100644 index 0000000..50c8efd --- /dev/null +++ b/internal/testinfqueueasmain_test.go @@ -0,0 +1,26 @@ +package main + +import ( + "testing" + + "github.com/jpaulm/gofbp" + "github.com/jpaulm/gofbp/io" + "github.com/jpaulm/gofbp/testrtn" +) + +func TestInfQueueAsMain(t *testing.T) { + net := gofbp.NewNetwork("InfQueueAsMain") + + proc1 := net.NewProc("Sender", &testrtn.Sender{}) + proc2 := net.NewProc("WriteFile", &io.WriteFile{}) + proc3 := net.NewProc("ReadFile", &io.ReadFile{}) + proc4 := net.NewProc("WriteToConsole", &testrtn.WriteToConsole{}) + + net.Initialize("40", proc1, "COUNT") + net.Initialize("testdata/infqueue", proc2, "FILENAME") + net.Initialize("testdata/infqueue", proc3, "FILENAME") + net.Connect(proc1, "OUT", proc2, "IN", 6) + net.Connect(proc2, "*", proc3, "*", 6) + net.Connect(proc3, "OUT", proc4, "IN", 6) + net.Run() +} diff --git a/testloadbal_test.go b/internal/testloadbal_test.go similarity index 83% rename from testloadbal_test.go rename to internal/testloadbal_test.go index 18998be..f5efd06 100644 --- a/testloadbal_test.go +++ b/internal/testloadbal_test.go @@ -3,12 +3,12 @@ package main import ( "testing" - "github.com/jpaulm/gofbp/components/testrtn" - "github.com/jpaulm/gofbp/core" + "github.com/jpaulm/gofbp" + "github.com/jpaulm/gofbp/testrtn" ) func TestLoadBal(t *testing.T) { - net := core.NewNetwork("TestLoadBal") + net := gofbp.NewNetwork("TestLoadBal") proc1 := net.NewProc("Sender", &testrtn.Sender{}) diff --git a/internal/testsubnet1_test.go b/internal/testsubnet1_test.go new file mode 100644 index 0000000..82031e0 --- /dev/null +++ b/internal/testsubnet1_test.go @@ -0,0 +1,25 @@ +package main + +import ( + "testing" + + "github.com/jpaulm/gofbp" + "github.com/jpaulm/gofbp/subnets" + "github.com/jpaulm/gofbp/testrtn" +) + +func TestSubnet1(t *testing.T) { + net := gofbp.NewNetwork("TestSubnet1") + + proc1 := net.NewProc("Sender1", &testrtn.Sender{}) + + proc2 := net.NewProc("RunSubnet", &subnets.Subnet1{}) + + proc3 := net.NewProc("WriteToConsole2", &testrtn.WriteToConsole{}) + + net.Initialize("15", proc1, "COUNT") + net.Connect(proc1, "OUT", proc2, "IN", 6) + net.Connect(proc2, "OUT", proc3, "IN", 6) + + net.Run() +} diff --git a/internal/testsubnet2_test.go b/internal/testsubnet2_test.go new file mode 100644 index 0000000..0a0da00 --- /dev/null +++ b/internal/testsubnet2_test.go @@ -0,0 +1,27 @@ +package main + +// Thiis differs from TestSubnet1 because it uses SSSubnet1 (SubstreamSenstive) instead of simple Subnet1... + +import ( + "testing" + + "github.com/jpaulm/gofbp" + "github.com/jpaulm/gofbp/subnets" + "github.com/jpaulm/gofbp/testrtn" +) + +func TestSubnet2(t *testing.T) { + net := gofbp.NewNetwork("TestSubnet2") + + proc1 := net.NewProc("SubstreamSender", &testrtn.SubstreamSender{}) // sends multiple substreams of varying lengths + + proc2 := net.NewProc("RunSSSubnet", &subnets.SSSubnet1{}) + + proc3 := net.NewProc("WriteToConsole2", &testrtn.WriteToConsole{}) + + net.Initialize("20", proc1, "COUNT") + net.Connect(proc1, "OUT", proc2, "IN", 6) + net.Connect(proc2, "OUT", proc3, "IN", 6) + + net.Run() +} diff --git a/internal/testsubnet3_test.go b/internal/testsubnet3_test.go new file mode 100644 index 0000000..37afec4 --- /dev/null +++ b/internal/testsubnet3_test.go @@ -0,0 +1,29 @@ +package main + +// Thiis differs from TestSubnet1 because it uses SSSubnet1 (SubstreamSenstive) instead of simple Subnet1... + +import ( + "testing" + + "github.com/jpaulm/gofbp" + "github.com/jpaulm/gofbp/subnets" + "github.com/jpaulm/gofbp/testrtn" +) + +func TestSubnet3(t *testing.T) { + net := gofbp.NewNetwork("TestSubnet3") + + proc1 := net.NewProc("SubstreamSender", &testrtn.SubstreamSender{}) // sends multiple substreams of varying lengths + proc1a := net.NewProc("WriteToConsole3", &testrtn.WriteToConsole{}) + + proc2 := net.NewProc("RunSSSubnet", &subnets.SSSubnet2{}) + + proc3 := net.NewProc("WriteToConsole2", &testrtn.WriteToConsole{}) + + net.Initialize("20", proc1, "COUNT") + net.Connect(proc1, "OUT", proc1a, "IN", 6) + net.Connect(proc1a, "OUT", proc2, "IN", 6) + net.Connect(proc2, "OUT", proc3, "IN", 6) + + net.Run() +} diff --git a/io/readfile.go b/io/readfile.go new file mode 100644 index 0000000..780ae23 --- /dev/null +++ b/io/readfile.go @@ -0,0 +1,70 @@ +package io + +import ( + //"core" + "fmt" + "io" + "os" + "strings" + + "github.com/jpaulm/gofbp" +) + +//ReadFile type defines ipt and opt +type ReadFile struct { + ipt gofbp.InputConn + opt gofbp.OutputConn +} + +//Setup method opens readFile +func (readFile *ReadFile) Setup(p *gofbp.Process) { + readFile.ipt = p.OpenInPort("FILENAME") + readFile.opt = p.OpenOutPort("OUT") +} + +//Execute method starts Process +func (readFile *ReadFile) Execute(p *gofbp.Process) { + + icpkt := p.Receive(readFile.ipt) + fname, ok := icpkt.Contents.(string) + if !ok { + panic("Parameter (file name) not a string") + } + f, err := os.Open(fname) + if err != nil { + panic("Unable to read file: " + fname) + } + p.Discard(icpkt) + p.Close(readFile.ipt) + + var pkt *gofbp.Packet + var rec string + + defer f.Close() + buf := make([]byte, 1024) + for { + n, err := f.Read(buf) + if err == io.EOF { + break + } + if err != nil { + fmt.Println(err) + continue + } + if n > 0 { + //fmt.Println(string(buf[:n])) + rec += string(buf[:n]) + } + } + + for { + i := strings.Index(rec, "\n") + if i == -1 { + break + } + pkt = p.Create(rec[:i]) + p.Send(readFile.opt, pkt) + rec = rec[i+1:] + } + +} diff --git a/io/writefile.go b/io/writefile.go new file mode 100644 index 0000000..23509a3 --- /dev/null +++ b/io/writefile.go @@ -0,0 +1,66 @@ +package io + +import ( + "fmt" + "os" + + "github.com/jpaulm/gofbp" +) + +// WriteFile type defines iptIP, ipt, and opt +type WriteFile struct { + iptIP gofbp.InputConn + ipt gofbp.InputConn + opt gofbp.OutputConn +} + +//Setup method initializes Process +func (writeFile *WriteFile) Setup(p *gofbp.Process) { + writeFile.iptIP = p.OpenInPort("FILENAME") + writeFile.ipt = p.OpenInPort("IN") + writeFile.opt = p.OpenOutPortOptional("OUT") +} + +//MustRun method +func (WriteFile) MustRun() {} + +//Execute method strts Process +func (writeFile *WriteFile) Execute(p *gofbp.Process) { + + icpkt := p.Receive(writeFile.iptIP) + fname, ok := icpkt.Contents.(string) + if !ok { + panic("Parameter (file name) not a string") + } + p.Discard(icpkt) + p.Close(writeFile.iptIP) + + f, err := os.Create(fname) + if err != nil { + panic("Unable to open file: " + fname) + } + defer f.Close() + + for { + var pkt = p.Receive(writeFile.ipt) + if pkt == nil { + break + } + + data := []byte(pkt.Contents.(string) + "\n") + + _, err2 := f.Write(data) + + if err2 != nil { + panic("Unable to write file: " + fname) + } + + if !writeFile.opt.IsConnected() { + p.Discard(pkt) + } else { + p.Send(writeFile.opt, pkt) + } + + } + fmt.Println(p.Name+": File", fname, "written") +} diff --git a/network.go b/network.go new file mode 100644 index 0000000..d7235d7 --- /dev/null +++ b/network.go @@ -0,0 +1,476 @@ +package gofbp + +import ( + "fmt" + "io" + "os" + "regexp" + "strconv" + "strings" + "sync" + "sync/atomic" +) + +//var pStack []*Process + +//var stkLevel int + +var tracing bool +var tracelocks bool +var generateGids bool + +type Network struct { + Name string + procs map[string]*Process + //conns map[string]inputCommon + wg sync.WaitGroup + mother *Process +} + +func NewNetwork(name string) *Network { + net := &Network{ + Name: name, + procs: make(map[string]*Process), + //conns: make(map[string]inputCommon), + wg: sync.WaitGroup{}, + } + + //stkLevel++ + //if stkLevel >= len(pStack) { + //pStack = append(pStack, nil) + //} + + return net +} + +func NewSubnet(Name string, p *Process) *Network { + net := &Network{ + Name: Name, + procs: make(map[string]*Process), + //conns: make(map[string]inputCommon), + wg: sync.WaitGroup{}, + } + + net.mother = p + return net +} + +func (n *Network) GetProc(nm string) *Process { + return n.procs[nm] +} + +func (n *Network) SetProc(p *Process, nm string) { + n.procs[nm] = p +} + +func LockTr(sc *sync.Cond, s string, p *Process) { + if tracelocks && p != nil { + fmt.Println(p.Name, s) + } + sc.L.Lock() +} + +func UnlockTr(sc *sync.Cond, s string, p *Process) { + sc.L.Unlock() + if tracelocks && p != nil { + fmt.Println(p.Name, s) + } +} + +func BdcastTr(sc *sync.Cond, s string, p *Process) { + sc.Broadcast() + if tracelocks && p != nil { + fmt.Println(p.Name, s) + } +} + +func WaitTr(sc *sync.Cond, s string, p *Process) { + if tracelocks && p != nil { + fmt.Println(p.Name, s) + } + sc.Wait() +} + +func trace(p *Process, s ...string) { + if tracing { + fmt.Print(p.Name, " "+strings.Trim(fmt.Sprint(s), "[]")+"\n") + } +} + +func traceNet(n *Network, s ...string) { + if tracing { + fmt.Print(n.Name, " "+strings.Trim(fmt.Sprint(s), "[]")+"\n") + } +} + +//func (n *Network) id() string { return fmt.Sprintf("%p", n) } + +func (n *Network) NewProc(nm string, comp Component) *Process { + + proc := &Process{ + Name: nm, + logFile: "", + component: comp, + status: Notstarted, + network: n, + } + //atomic.StoreInt32(&proc.status, Notstarted) + n.SetProc(proc, nm) + proc.inPorts = make(map[string]inputCommon) + proc.outPorts = make(map[string]outputCommon) + proc.mtx = sync.Mutex{} + proc.canGo = sync.NewCond(&proc.mtx) + //if stkLevel > 0 { + // n.mother = pStack[stkLevel-1] + //} + proc.gid = getGID() + return proc +} + +func (n *Network) NewConnection(cap int) *InPort { + conn := &InPort{ + network: n, + } + conn.mtx = sync.Mutex{} + conn.condNE = sync.NewCond(&conn.mtx) + conn.condNF = sync.NewCond(&conn.mtx) + conn.pktArray = make([]*Packet, cap) + return conn +} + +func (n *Network) NewInitializationConnection() *InitializationConnection { + conn := &InitializationConnection{ + network: n, + } + conn.mtx = sync.Mutex{} + return conn +} + +func (n *Network) NewInArrayPort() *InArrayPort { + conn := &InArrayPort{ + network: n, + } + conn.mtx = sync.Mutex{} + return conn +} + +func (n *Network) NewOutArrayPort() *OutArrayPort { + port := &OutArrayPort{ + network: n, + } + + return port +} + +func (n *Network) Connect(p1 *Process, out string, p2 *Process, in string, cap int) { + + inPort := parsePort(in) + + var connxn *InPort + //var anyInConn InputConn + + if inPort.indexed { + var anyInConn = p2.inPorts[inPort.name] + if anyInConn == nil { + + anyInConn = n.NewInArrayPort() + p2.inPorts[inPort.name] = anyInConn + } + + connxn = anyInConn.(InputArrayConn).GetArrayItem(inPort.index) + + if connxn == nil { + connxn = n.NewConnection(cap) + connxn.portName = inPort.name + "[" + strconv.Itoa(inPort.index) + "]" + connxn.fullName = p2.Name + "." + connxn.portName + //n.conns[connxn.fullName] = connxn + connxn.downStrProc = p2 + connxn.network = n + if anyInConn == nil { + p2.inPorts[inPort.name] = connxn + } else { + anyInConn.(InputArrayConn).SetArrayItem(connxn, inPort.index) + } + } + } else { + if p2.inPorts[inPort.name] == nil { + connxn = n.NewConnection(cap) + connxn.portName = inPort.name + connxn.fullName = p2.Name + "." + inPort.name + //n.conns[connxn.fullName] = connxn + connxn.downStrProc = p2 + connxn.network = n + p2.inPorts[inPort.name] = connxn + } else { + connxn = p2.inPorts[inPort.name].(*InPort) + } + } + + if in == "*" { + p2.autoInput = connxn + } + + // connxn built; input port array built if necessary + + outPort := parsePort(out) + + var opt *OutPort + + if outPort.indexed { + var anyOutConn = p1.outPorts[outPort.name] + if anyOutConn == nil { + anyOutConn = n.NewOutArrayPort() + p1.outPorts[outPort.name] = anyOutConn + } + + //opt := new(OutArrayPort) + outConn := anyOutConn.(*OutArrayPort) + //p1.outPorts[out] = anyOutConn + //opt.Name = out + opt = new(OutPort) + outConn.SetArrayItem(opt, outPort.index) + opt.portName = out + opt.fullName = p1.Name + "." + out + //n.conns[opt.fullName] = opt.Conn + } else { + //var opt OutputConn + opt = new(OutPort) + p1.outPorts[out] = opt + opt.network = n + opt.portName = out + opt.fullName = p1.Name + "." + out + //n.conns[opt.fullName] = opt.Conn + + } + + opt.sender = p1 + opt.Conn = connxn + opt.connected = true + + connxn.incUpstream() + if out == "*" { + p1.autoOutput = connxn + } +} + +type portDefinition struct { + name string + index int + indexed bool +} + +var portPattern = regexp.MustCompile(`^(.+)\[(\d+)\]$`) + +func parsePort(in string) portDefinition { + matches := portPattern.FindStringSubmatch(in) + if len(matches) == 0 { + return portDefinition{name: in} + } + root, indexStr := matches[1], matches[2] + + index, err := strconv.Atoi(indexStr) + if err != nil { + panic("Invalid index in " + in) + } + + return portDefinition{name: root, index: index, indexed: true} +} + +func (n *Network) Initialize(initValue interface{}, p2 *Process, in string) { + + conn := n.NewInitializationConnection() + p2.inPorts[in] = conn + conn.portName = in + conn.fullName = p2.Name + "." + in + //n.conns[conn.fullName] = conn + + conn.value = initValue +} + +func (n *Network) Exit() { + if n.mother == nil { + traceNet(n, "Exit network") + } else { + traceNet(n, "Exit subnet") + } +} + +func setOptions() { + var rec string + f, err := os.Open("params.xml") + if err == nil { + defer f.Close() + buf := make([]byte, 1024) + for { + n, err := f.Read(buf) + if err == io.EOF { + break + } + if err != nil { + fmt.Println(err) + continue + } + if n > 0 { + //fmt.Println(string(buf[:n])) + rec += string(buf[:n]) + } + } + + s := "" + i := strings.Index(rec, s) + if i > -1 && rec[i+len(s):i+len(s)+4] == "true" { + tracing = true + } + + s = "" + i = strings.Index(rec, s) + if i > -1 && rec[i+len(s):i+len(s)+4] == "true" { + tracelocks = true + } + + s = "" + i = strings.Index(rec, s) + if i > -1 && rec[i+len(s):i+len(s)+4] == "true" { + generateGids = true + } + } +} + +func (n *Network) Run() { + defer n.Exit() + if n.mother == nil { + setOptions() + } + + defer traceNet(n, " Done") + + for { + if n.mother != nil { + traceNet(n, " Starting subnet activation") + } else { + traceNet(n, " Starting network") + } + + // FBP distinguishes between execution of the process as a whole and activating the code - the code may be deactivated and then + // reactivated many times during the process "run" + + n.wg = sync.WaitGroup{} + n.wg.Add(len(n.procs)) + + //defer n.wg.Wait() + var someProcsCanRun bool = false + + for _, proc := range n.procs { + + //LockTr(proc.canGo, "test if not started L", proc) + //atomic.StoreInt32(&proc.status, Notstarted) + + selfStarting := true + if proc.inPorts != nil { + for _, conn := range proc.inPorts { + + _, b := conn.(*InitializationConnection) + if !b { + selfStarting = false + } + } + } + + if !selfStarting { + continue + } + //atomic.StoreInt32(&proc.status, Notstarted) + trace(proc, "act from start") + proc.activate() + someProcsCanRun = true + //UnlockTr(proc.canGo, "test if not started U", proc) + } + if !someProcsCanRun { + n.wg.Add(0 - len(n.procs)) + panic("No process can start") + } + n.wg.Wait() + + if n.mother == nil { + return + } + + traceNet(n, " subnet deactivated") + + for _, p := range n.procs { + for _, v := range p.inPorts { + _, b := v.(*InArrayPort) + if b { + for _, w := range v.(*InArrayPort).array { + w.resetForNextExecution() + } + } else { + v.resetForNextExecution() + } + } + } + + var p *Process + //if n.mother != nil { + p = n.mother + //} else { + // return + //} + + allDrained, _, _ := p.inputState() + if allDrained { + break + } + + for _, p := range n.procs { + atomic.StoreInt32(&p.status, Notstarted) + for _, v := range p.outPorts { + _, b := v.(*OutArrayPort) + if b { + for _, w := range v.(*OutArrayPort).array { + w.Conn.upStrmCnt = 0 + } + } else { + w, b := v.(*OutPort) + if b { + w.Conn.upStrmCnt = 0 + } + } + } + } + + for _, p := range n.procs { + for _, v := range p.outPorts { + _, b := v.(*OutArrayPort) + if b { + for _, w := range v.(*OutArrayPort).array { + w.Conn.incUpstream() + } + } else { + w, b := v.(*OutPort) + if b { + w.Conn.incUpstream() + } + } + } + } + } + + //if n.mother != nil { + p := n.mother + for _, v := range p.outPorts { + _, b := v.(*OutArrayPort) + if b { + for _, w := range v.(*OutArrayPort).array { + w.Close() + } + } else { + w, b := v.(*OutPort) + if b { + w.Close() + } + } + } +} + +//} diff --git a/nulloutport.go b/nulloutport.go new file mode 100644 index 0000000..9d31c6c --- /dev/null +++ b/nulloutport.go @@ -0,0 +1,19 @@ +package gofbp + +type NullOutPort struct { + sender *Process + portName string + fullName string +} + +// NullOutPort by default discards the packet. +func (o *NullOutPort) send(p *Process, pkt *Packet) bool { + p.Discard(pkt) + return true +} + +func (o *NullOutPort) IsConnected() bool { + return false +} + +func (o *NullOutPort) Close() {} diff --git a/outarrayport.go b/outarrayport.go new file mode 100644 index 0000000..c6154cf --- /dev/null +++ b/outarrayport.go @@ -0,0 +1,64 @@ +package gofbp + +import ( + //"fmt" + "math" +) + +type OutArrayPort struct { + network *Network + + portName string + fullName string + array []*OutPort + //closed bool + connected bool + //sender *Process +} + +func (o *OutArrayPort) send(p *Process, pkt *Packet) bool { panic("send on array port") } + +func (o *OutArrayPort) GetArrayItem(i int) *OutPort { + if i >= len(o.array) { + return nil + } + return o.array[i] +} + +func (o *OutArrayPort) SetArrayItem(o2 *OutPort, i int) { + if i >= len(o.array) { + // add to .array to fit c2 + increaseBy := make([]*OutPort, i-len(o.array)+1) + o.array = append(o.array, increaseBy...) + } + o.array[i] = o2 +} + +func (o *OutArrayPort) ArrayLength() int { + return len(o.array) +} + +func (o *OutArrayPort) Close() { + for _, v := range o.array { + v.Close() + } +} + +func (o *OutArrayPort) IsConnected() bool { + //return o.connected + return true +} + +func (o *OutArrayPort) GetItemWithFewestIPs() int { + backlog := math.MaxInt32 + var fewestIPsIndex int + for i, v := range o.array { + j := v.Conn.PktCount() + //fmt.Println(i, j) + if j <= backlog { + fewestIPsIndex = i + backlog = j + } + } + return fewestIPsIndex +} diff --git a/outport.go b/outport.go new file mode 100644 index 0000000..6be612d --- /dev/null +++ b/outport.go @@ -0,0 +1,94 @@ +package core + +import ( + "fmt" + "sync/atomic" +) + +type OutPort struct { + portName string + fullName string + conn *InPort + connected bool + sender *Process + network *Network +} + +func (o *OutPort) send(p *Process, pkt *Packet) bool { + if o == nil { + return false + } + + if pkt == nil { + panic("Sending nil packet") + } + if pkt.owner != p { + panic("Sending packet not owned by this process") + } + + LockTr(o.conn.condNF, "send L", p) + defer UnlockTr(o.conn.condNF, "send U", p) + + if pkt.PktType != NormalPacket { + trace(p, " Sending to "+o.portName+" > "+ + [...]string{"", "Open", "Close"}[pkt.PktType]+" Bracket") + if tracing { + fmt.Print(" contents: ", pkt.Contents, "\n") + } + } else { + trace(p, " Sending to "+o.portName+" > ") + if tracing { + fmt.Print(" ", pkt.Contents, "\n") + } + } + + for o.conn.isFull() { // while connection is full + atomic.StoreInt32(&p.status, SuspSend) + WaitTr(o.conn.condNF, "wait in send", p) + atomic.StoreInt32(&p.status, Active) + } + + trace(p, " Sent to "+o.portName) + + trace(o.conn.downStrProc, "act from send") + o.conn.downStrProc.activate() + + o.conn.pktArray[o.conn.is] = pkt + o.conn.is = (o.conn.is + 1) % len(o.conn.pktArray) + //pkt.owner = nil + p.ownedPkts-- + pkt = nil + BdcastTr(o.conn.condNE, "bdcast sent", p) + return true +} + +func (o *OutPort) IsConnected() bool { + if o == nil { + return false + } + return o.connected +} + +func (o *OutPort) GetArrayItem(i int) *OutPort { + return nil +} + +func (o *OutPort) SetArrayItem(op *OutPort, i int) {} + +func (o *OutPort) ArrayLength() int { + return 0 +} + +func (o *OutPort) Close() { + LockTr(o.conn.condNF, "close L", o.sender) + defer UnlockTr(o.conn.condNF, "close U", o.sender) + trace(o.sender, " Close "+o.portName) + + o.conn.decUpstream() + if o.conn.upStrmCnt == 0 { + o.conn.closed = true + BdcastTr(o.conn.condNE, "bdcast out", o.sender) + trace(o.conn.downStrProc, "activated from close") + o.conn.downStrProc.activate() + } +} diff --git a/outputconn.go b/outputconn.go new file mode 100644 index 0000000..e91aa7c --- /dev/null +++ b/outputconn.go @@ -0,0 +1,20 @@ +package gofbp + +type outputCommon interface { + Close() + IsConnected() bool +} + +type OutputConn interface { + outputCommon + send(*Process, *Packet) bool + // IsConnected() bool +} + +type OutputArrayConn interface { + outputCommon + GetArrayItem(i int) *OutPort + SetArrayItem(c *OutPort, i int) + ArrayLength() int + GetItemWithFewestIPs() int +} diff --git a/packet.go b/packet.go new file mode 100644 index 0000000..8a12c62 --- /dev/null +++ b/packet.go @@ -0,0 +1,13 @@ +package gofbp + +const ( + Normal int32 = iota + OpenBracket + CloseBracket +) + +type Packet struct { + Contents interface{} + PktType int32 + owner *Process +} diff --git a/params.xml b/params.xml index 2bf5e9b..80244b7 100644 --- a/params.xml +++ b/params.xml @@ -1,6 +1,7 @@ - + -false -false -false + true + false + true + false diff --git a/pkg/mod/cache/download/golang.org/x/tools/gopls/@v/list b/pkg/mod/cache/download/golang.org/x/tools/gopls/@v/list deleted file mode 100644 index 3d105a6..0000000 --- a/pkg/mod/cache/download/golang.org/x/tools/gopls/@v/list +++ /dev/null @@ -1 +0,0 @@ -v0.7.3 diff --git a/pkg/mod/cache/download/golang.org/x/tools/gopls/@v/v0.7.3.info b/pkg/mod/cache/download/golang.org/x/tools/gopls/@v/v0.7.3.info deleted file mode 100644 index e5ac8d4..0000000 --- a/pkg/mod/cache/download/golang.org/x/tools/gopls/@v/v0.7.3.info +++ /dev/null @@ -1 +0,0 @@ -{"Version":"v0.7.3","Time":"2021-10-15T15:11:15Z"} \ No newline at end of file diff --git a/pkg/mod/cache/download/golang.org/x/tools/gopls/@v/v0.7.3.mod b/pkg/mod/cache/download/golang.org/x/tools/gopls/@v/v0.7.3.mod deleted file mode 100644 index 795c149..0000000 --- a/pkg/mod/cache/download/golang.org/x/tools/gopls/@v/v0.7.3.mod +++ /dev/null @@ -1,24 +0,0 @@ -module golang.org/x/tools/gopls - -go 1.18 - -require ( - github.com/google/go-cmp v0.5.6 - github.com/jba/templatecheck v0.6.0 - github.com/sanity-io/litter v1.5.1 - github.com/sergi/go-diff v1.1.0 - golang.org/x/mod v0.4.2 - golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e - golang.org/x/tools v0.1.8-0.20211014194737-fc98fb2abd48 - honnef.co/go/tools v0.2.0 - mvdan.cc/gofumpt v0.1.1 - mvdan.cc/xurls/v2 v2.3.0 -) - -require ( - github.com/BurntSushi/toml v0.3.1 // indirect - github.com/google/safehtml v0.0.2 // indirect - golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect - golang.org/x/text v0.3.6 // indirect - golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect -) diff --git a/pkg/mod/cache/download/sumdb/sum.golang.org/lookup/golang.org/x/tools/gopls@v0.7.3 b/pkg/mod/cache/download/sumdb/sum.golang.org/lookup/golang.org/x/tools/gopls@v0.7.3 deleted file mode 100644 index 719a1d7..0000000 --- a/pkg/mod/cache/download/sumdb/sum.golang.org/lookup/golang.org/x/tools/gopls@v0.7.3 +++ /dev/null @@ -1,9 +0,0 @@ -7350634 -golang.org/x/tools/gopls v0.7.3 h1:Lru57ht8vtDMouRskFC085VAjBAZRAISd/lwvwOOV0Q= -golang.org/x/tools/gopls v0.7.3/go.mod h1:I2g1ow+mLbhf12iRXU2CMngcTXz8ge7kbrQUNp34Axw= - -go.sum database tree -7631008 -mdir6bwUx2zzU+4xJ7mpBZ6jby5m3oZ6dkmNRClHxBI= - -— sum.golang.org Az3griw85MEInWHLd6r3nZnTlE8xL4ctOg1Q0gNRc0op67cpiVp3sWOB04qZEL9QxyfTcpdy2mKfvYXsKMZRwKfSdQE= diff --git a/pkg/mod/cache/download/sumdb/sum.golang.org/tile/8/0/x028/713 b/pkg/mod/cache/download/sumdb/sum.golang.org/tile/8/0/x028/713 deleted file mode 100644 index e1d4eec..0000000 Binary files a/pkg/mod/cache/download/sumdb/sum.golang.org/tile/8/0/x028/713 and /dev/null differ diff --git a/pkg/mod/cache/download/sumdb/sum.golang.org/tile/8/0/x029/808.p/160 b/pkg/mod/cache/download/sumdb/sum.golang.org/tile/8/0/x029/808.p/160 deleted file mode 100644 index 071b0ea..0000000 Binary files a/pkg/mod/cache/download/sumdb/sum.golang.org/tile/8/0/x029/808.p/160 and /dev/null differ diff --git a/pkg/mod/cache/download/sumdb/sum.golang.org/tile/8/1/112 b/pkg/mod/cache/download/sumdb/sum.golang.org/tile/8/1/112 deleted file mode 100644 index 7dc6d78..0000000 Binary files a/pkg/mod/cache/download/sumdb/sum.golang.org/tile/8/1/112 and /dev/null differ diff --git a/pkg/mod/cache/download/sumdb/sum.golang.org/tile/8/1/116.p/112 b/pkg/mod/cache/download/sumdb/sum.golang.org/tile/8/1/116.p/112 deleted file mode 100644 index 1cbc3a4..0000000 Binary files a/pkg/mod/cache/download/sumdb/sum.golang.org/tile/8/1/116.p/112 and /dev/null differ diff --git a/pkg/mod/cache/download/sumdb/sum.golang.org/tile/8/2/000.p/116 b/pkg/mod/cache/download/sumdb/sum.golang.org/tile/8/2/000.p/116 deleted file mode 100644 index 7feeebf..0000000 Binary files a/pkg/mod/cache/download/sumdb/sum.golang.org/tile/8/2/000.p/116 and /dev/null differ diff --git a/pkg/mod/cache/lock b/pkg/mod/cache/lock deleted file mode 100644 index e69de29..0000000 diff --git a/pkg/sumdb/sum.golang.org/latest b/pkg/sumdb/sum.golang.org/latest deleted file mode 100644 index 30e15ad..0000000 --- a/pkg/sumdb/sum.golang.org/latest +++ /dev/null @@ -1,5 +0,0 @@ -go.sum database tree -7631008 -mdir6bwUx2zzU+4xJ7mpBZ6jby5m3oZ6dkmNRClHxBI= - -— sum.golang.org Az3griw85MEInWHLd6r3nZnTlE8xL4ctOg1Q0gNRc0op67cpiVp3sWOB04qZEL9QxyfTcpdy2mKfvYXsKMZRwKfSdQE= diff --git a/process.go b/process.go new file mode 100644 index 0000000..7e80081 --- /dev/null +++ b/process.go @@ -0,0 +1,332 @@ +package gofbp + +import ( + "bytes" + "fmt" + "runtime" + "strconv" + "sync" + "sync/atomic" +) + +//Process type defines a gofbp process. +type Process struct { + gid uint64 + Name string + network *Network + + inPorts map[string]inputCommon + outPorts map[string]outputCommon + + logFile string + component Component + ownedPkts int + status int32 + mtx sync.Mutex + canGo *sync.Cond + autoInput inputCommon + autoOutput inputCommon +} + +//Nostarted const reflects current process status +const ( + Notstarted int32 = iota + Active + Dormant + SuspSend + SuspRecv + Terminated +) + +//OpenInPort function opens and returns InputConn +func (p *Process) OpenInPort(s string) InputConn { + var in InputConn + var b bool + if len(p.inPorts) == 0 { + panic(p.Name + ": No input ports specified") + } + in, b = p.inPorts[s].(InputConn) + + if !b { + panic(p.Name + " " + s + " InPort not connected, or found other type") + } + return in +} + +//OpenInArrayPort method opens and returns InArrayPort +func (p *Process) OpenInArrayPort(s string) *InArrayPort { + var in *InArrayPort + var b bool + if len(p.inPorts) == 0 { + panic(p.Name + ": No input ports specified") + } + in, b = p.inPorts[s].(*InArrayPort) + //if in == nil { + // panic(p.Name + ": Port Name not found (" + s + ")") + //} + if !b { + panic(p.Name + " " + s + " InArrayPort not connected, or found other type") + } + return in +} + +//OpenOutPort method opens and returns OutputConn +func (p *Process) OpenOutPort(s string) OutputConn { + var out OutputConn + var b bool + if len(p.outPorts) == 0 { + panic(p.Name + " " + s + " OutPort not connected") + } else { + out, b = p.outPorts[s].(*OutPort) + if !b { + panic(p.Name + " " + s + " OutPort not connected, or found other type") + } + out.(*OutPort).portName = s + out.(*OutPort).fullName = p.Name + "." + s + //p.network.conns[out.(*OutPort).fullName] = out.(*OutPort).Conn + } + + return out + +} + +//OpenOutPortOptional function opens and returns OutputConn +func (p *Process) OpenOutPortOptional(s string) OutputConn { + var out OutputConn + var b bool + if len(p.outPorts) == 0 { + out = new(NullOutPort) + p.outPorts[s] = out + } else { + out, b = p.outPorts[s].(*OutPort) + if b { + out.(*OutPort).portName = s + out.(*OutPort).fullName = p.Name + "." + s + //p.network.conns[out.(*OutPort).fullName] = out.(*OutPort).Conn + } else { + out := new(NullOutPort) + p.outPorts[s] = out + } + } + + return out +} + +// not sure it makes sense to allow optional for array ports! + +//OpenOutArrayPort method opens and returns OutArrayPort +func (p *Process) OpenOutArrayPort(s string) *OutArrayPort { + var out *OutArrayPort + var b bool + if len(p.outPorts) == 0 { + out = new(OutArrayPort) + p.outPorts[s] = out + out.portName = s + out.fullName = p.Name + "." + s + out.connected = false + } else { + out, b = p.outPorts[s].(*OutArrayPort) + if !b { + panic(p.Name + " " + s + " OutArrayPort not connected, or found other type") + } + } + + return out +} + +// Send sends a packet to the output connection. + +//Send method emits Packet, returning false when fails to send. +func (p *Process) Send(o OutputConn, pkt *Packet) bool { + //o.SetSender(p) + return o.send(p, pkt) +} + +//Receive method accepts InputConn and returns Packet +// Receive receives from the connection. +// Returns nil, when there's no more data. +func (p *Process) Receive(c InputConn) *Packet { + return c.receive(p) +} + +//Close method closes InputConn +func (p *Process) Close(c InputConn) { + c.Close() +} + +func (p *Process) activate() { + + // + // This function starts a goroutine if it is not started, and signal if it has been + // + + //LockTr(p.canGo, "start test L", p) + //trace(p, "test if notstarted - status: "+strconv.FormatInt(int64(p.status), 10)) + if !atomic.CompareAndSwapInt32(&p.status, Notstarted, Active) { + if atomic.CompareAndSwapInt32(&p.status, Dormant, Active) { + + BdcastTr(p.canGo, "bdcast act", p) + } + //UnlockTr(p.canGo, "start test U", p) + return + } + //UnlockTr(p.canGo, "start test U", p) + + go func() { // Process goroutine + defer p.network.wg.Done() + trace(p, "Starting goroutine "+strconv.FormatUint(getGID(), 10)) + p.Run() // <------- + }() + +} + +func (p *Process) inputState() (bool, bool, bool) { + + LockTr(p.canGo, "IS L", p) + defer UnlockTr(p.canGo, "IS U", p) + + allDrained := true + hasData := false + selfStarting := true + + for { + for _, v := range p.inPorts { + _, b := v.(*InArrayPort) + if b { + for _, w := range v.(*InArrayPort).array { + allDrained = allDrained && w.IsDrained() + hasData = hasData || !w.IsEmpty() + selfStarting = false + } + } else { + w, b := v.(*InPort) + if b { + allDrained = allDrained && v.IsDrained() + hasData = hasData || !w.IsEmpty() + selfStarting = false + } + } + } + + if allDrained || hasData || selfStarting { + return allDrained, hasData, selfStarting + } + + atomic.StoreInt32(&p.status, Dormant) + WaitTr(p.canGo, "wait in IS", p) + + } +} + +//Run method initializes and Executes Process +func (p *Process) Run() { + + defer atomic.StoreInt32(&p.status, Terminated) + defer trace(p, " terminated") + trace(p, " started") + + if generateGids { + fmt.Println("Goroutine", p.Name+":", "no.", getGID()) + } + + p.component.Setup(p) + + //if p.selfStarting { + // autoStarting = true + //} + allDrained, hasData, selfStarting := p.inputState() + + canRun := selfStarting || hasData || !allDrained || p.autoInput != nil || p.isMustRun() && allDrained + + for canRun { + + // multiple activations, if necessary! + + trace(p, " activated") + atomic.StoreInt32(&p.status, Active) + + p.component.Execute(p) // single "activation" + + atomic.StoreInt32(&p.status, Dormant) + trace(p, " deactivated") + + if p.ownedPkts > 0 { + panic(p.Name + " deactivated without disposing of all owned packets") + } + + if selfStarting { + break + } + + allDrained, _, _ := p.inputState() + + if allDrained { + if p.autoOutput != nil { + p.autoOutput.Close() + } + break + } + + for _, v := range p.inPorts { + + _, b := v.(*InitializationConnection) + if b { + v.resetForNextExecution() + } + } + } + + for _, v := range p.outPorts { + + if v.IsConnected() { + v.Close() + } + } + +} + +func (p *Process) isMustRun() bool { + _, hasMustRun := p.component.(ComponentWithMustRun) + return hasMustRun +} + +//Create method creates and interface and returns a Packet +// create packet containing anything! +func (p *Process) Create(x interface{}) *Packet { + var pkt *Packet = new(Packet) + pkt.Contents = x + pkt.owner = p + p.ownedPkts++ + return pkt +} + +//CreateBracket method builds a new Packet and returns it +// create bracket +func (p *Process) CreateBracket(pktType int32, s string) *Packet { + var pkt *Packet = new(Packet) + pkt.Contents = s + pkt.PktType = pktType + pkt.owner = p + p.ownedPkts++ + return pkt +} + +//Discard method sets Packet to nil +func (p *Process) Discard(pkt *Packet) { + if pkt == nil { + panic("Discarding nil packet") + } + p.ownedPkts-- + pkt = nil +} + +//https://blog.sgmansfield.com/2015/12/goroutine-ids/ + +func getGID() uint64 { + b := make([]byte, 64) + b = b[:runtime.Stack(b, false)] + b = bytes.TrimPrefix(b, []byte("goroutine ")) + b = b[:bytes.IndexByte(b, ' ')] + n, _ := strconv.ParseUint(string(b), 10, 64) + return n +} diff --git a/scripts/chat2.html b/scripts/chat2.html new file mode 100644 index 0000000..27a263a --- /dev/null +++ b/scripts/chat2.html @@ -0,0 +1,169 @@ + + + +WebSocket Example - Chat2 + +// + + + + +

WebSockets Example - Chat2

+ +

+ Command: +

+

+ Data: +

+

+ Status: +

+ + +
    +
+ + + + diff --git a/settings.json b/settings.json index fb15a9c..63e4199 100644 --- a/settings.json +++ b/settings.json @@ -1,3 +1,4 @@ { "go.testFlags": ["-race"] -} \ No newline at end of file + + } \ No newline at end of file diff --git a/subnet_support.go b/subnet_support.go new file mode 100644 index 0000000..212ba74 --- /dev/null +++ b/subnet_support.go @@ -0,0 +1,168 @@ +package gofbp + +import ( + //"fmt" + "strings" +) + +type SubIn struct { + //ipt InputConn + iptIP InputConn + out OutputConn + eipt InputConn +} + +type SubInSS struct { + SubIn +} + +func (subIn *SubIn) Setup(p *Process) { + + //SubIn.ipt = p.OpenInPort("IN") + + subIn.out = p.OpenOutPort("OUT") + + subIn.iptIP = p.OpenInPort("NAME") +} + +func (subIn *SubIn) Execute(p *Process) { + + icpkt := p.Receive(subIn.iptIP) + param := icpkt.Contents.(string) + + p.Discard(icpkt) + + mother := p.network.mother + subIn.eipt = mother.OpenInPort(param) + if !strings.Contains(subIn.eipt.(*InPort).portName, ".") { + subIn.eipt.(*InPort).portName = mother.Name + ":" + subIn.eipt.(*InPort).portName + } + + for { + //var pkt = mother.Receive(subIn.eipt) + var pkt = subIn.eipt.receive(p) + if pkt == nil { + //trace(p, " Received end of stream") + break + } + + pkt.owner = p + //fmt.Println(pkt.Contents) + p.Send(subIn.out, pkt) + + } + +} + +func (subIn *SubInSS) Execute(p *Process) { + + icpkt := p.Receive(subIn.iptIP) + param := icpkt.Contents.(string) + + p.Discard(icpkt) + mother := p.network.mother + subIn.eipt = mother.OpenInPort(param) + if !strings.Contains(subIn.eipt.(*InPort).portName, ":") { + subIn.eipt.(*InPort).portName = mother.Name + ":" + subIn.eipt.(*InPort).portName + } + + for { + //var pkt = mother.Receive(subIn.eipt) + var pkt = subIn.eipt.receive(p) + if pkt == nil { + //trace(p, " Received end of stream") + break + } + + if pkt.PktType == OpenBracket { + p.Discard(pkt) + } else { + if pkt.PktType != CloseBracket { + pkt.owner = p + //fmt.Println(pkt.Contents) + p.Send(subIn.out, pkt) + } else { + p.Discard(pkt) + //subIn.out.Close() + return + } + } + } +} + +type SubOut struct { + ipt InputConn + iptIP InputConn + eopt OutputConn +} + +type SubOutSS struct { + SubOut +} + +func (subOut *SubOut) Setup(p *Process) { + + subOut.ipt = p.OpenInPort("IN") + + subOut.iptIP = p.OpenInPort("NAME") +} + +//func (SubOut) MustRun() {} + +func (subOut *SubOut) Execute(p *Process) { + + icpkt := p.Receive(subOut.iptIP) + param := icpkt.Contents.(string) + + p.Discard(icpkt) + mother := p.network.mother + subOut.eopt = mother.OpenOutPort(param) + if !strings.Contains(subOut.eopt.(*OutPort).portName, ":") { + subOut.eopt.(*OutPort).portName = mother.Name + ":" + subOut.eopt.(*OutPort).portName + } + + for { + + pkt := subOut.ipt.receive(p) + if pkt == nil { + //trace(p, " Received end of stream") + break + } + pkt.owner = p + + //fmt.Println(pkt.Contents) + + subOut.eopt.send(p, pkt) + } +} +func (subOut *SubOutSS) Execute(p *Process) { + + icpkt := p.Receive(subOut.iptIP) + param := icpkt.Contents.(string) + + p.Discard(icpkt) + mother := p.network.mother + subOut.eopt = mother.OpenOutPort(param) + if !strings.Contains(subOut.eopt.(*OutPort).portName, ":") { + subOut.eopt.(*OutPort).portName = mother.Name + ":" + subOut.eopt.(*OutPort).portName + } + + pkt := p.CreateBracket(OpenBracket, "") + subOut.eopt.send(p, pkt) + + for { + pkt = subOut.ipt.receive(p) + if pkt == nil { + //trace(p, " Received end of stream") + break + } + pkt.owner = p + + //fmt.Println(pkt.Contents) + + subOut.eopt.send(p, pkt) + + } + pkt = p.CreateBracket(CloseBracket, "") + subOut.eopt.send(p, pkt) +} diff --git a/subnet_tests/subnet-diagram.go b/subnet_tests/subnet-diagram.go new file mode 100644 index 0000000..175453c --- /dev/null +++ b/subnet_tests/subnet-diagram.go @@ -0,0 +1,20 @@ +package main //change package name, or delete statement, if desired + +// Network generated by DrawFBP + +import ( + "github.com/jpaulm/gofbp/components/subnets" + "github.com/jpaulm/gofbp/components/testrtn" + "github.com/jpaulm/gofbp/core" +) + +func main() { + net := core.NewNetwork("Subnet-diagram") + substream___sender := net.NewProc("Substream___Sender", &testrtn.Sender{}) + run___subnet := net.NewProc("Run___Subnet", &subnets.Subnet1{}) + write___to_console := net.NewProc("Write___To_Console", &testrtn.WriteToConsole{}) + net.Connect(substream___sender, "OUT", run___subnet, "IN", 1) + net.Connect(run___subnet, "OUT", write___to_console, "IN", 1) + net.Initialize("40", substream___sender, "COUNT") + net.Run() +} diff --git a/subnet_tests/subnets/subnet1.go b/subnet_tests/subnets/subnet1.go new file mode 100644 index 0000000..9572c1d --- /dev/null +++ b/subnet_tests/subnets/subnet1.go @@ -0,0 +1,26 @@ +package subnets //change package name, or delete statement, if desired + +// Network generated by DrawFBP + +import ( + "github.com/jpaulm/gofbp/components/testrtn" + "github.com/jpaulm/gofbp/core" +) + +type Subnet1 struct { +} + +func (subnet *Subnet1) Setup(p *core.Process) { +} + +func (subnet *Subnet1) Execute(p *core.Process) { + net := core.NewSubnet("Subnet1", p) + write____to_console := net.NewProc("Write____To_Console", &testrtn.WriteToConsNL{}) + subin1 := net.NewProc("SUBIN1", &core.SubInSS{}) + net.Initialize("IN", subin1, "NAME") + subout2 := net.NewProc("SUBOUT2", &core.SubOutSS{}) + net.Initialize("OUT", subout2, "NAME") + net.Connect(subin1, "OUT", write____to_console, "IN", 6) + net.Connect(write____to_console, "OUT", subout2, "IN", 6) + net.Run() +} diff --git a/subnets/sssubnet1.go b/subnets/sssubnet1.go new file mode 100644 index 0000000..003a2b8 --- /dev/null +++ b/subnets/sssubnet1.go @@ -0,0 +1,27 @@ +package subnets + +import ( + "github.com/jpaulm/gofbp" + "github.com/jpaulm/gofbp/testrtn" +) + +type SSSubnet1 struct{} + +func (subnet *SSSubnet1) Setup(p *gofbp.Process) {} + +func (subnet *SSSubnet1) Execute(p *gofbp.Process) { + net := gofbp.NewSubnet("SSSubnet1", p) + + proc1 := net.NewProc("SubInSS", &gofbp.SubInSS{}) // Substream-Sensitive SubIn + + proc2 := net.NewProc("WriteToConsole1", &testrtn.WriteToConsole{}) + + proc3 := net.NewProc("SubOutSS", &gofbp.SubOutSS{}) // Substream-delimiter-Generating SubOut + + net.Initialize("IN", proc1, "NAME") + net.Connect(proc1, "OUT", proc2, "IN", 6) + net.Connect(proc2, "OUT", proc3, "IN", 6) + net.Initialize("OUT", proc3, "NAME") + + net.Run() +} diff --git a/subnets/sssubnet2.go b/subnets/sssubnet2.go new file mode 100644 index 0000000..3506785 --- /dev/null +++ b/subnets/sssubnet2.go @@ -0,0 +1,27 @@ +package subnets + +import ( + "github.com/jpaulm/gofbp" + "github.com/jpaulm/gofbp/testrtn" +) + +type SSSubnet2 struct{} + +func (subnet *SSSubnet2) Setup(p *gofbp.Process) {} + +func (subnet *SSSubnet2) Execute(p *gofbp.Process) { + net := gofbp.NewSubnet("SSSubnet2", p) + + proc1 := net.NewProc("SubInSS", &gofbp.SubInSS{}) // Substream-Sensitive SubIn + + proc2 := net.NewProc("Count", &testrtn.Counter{}) // count length of each substream (excl. brackets) + + proc3 := net.NewProc("SubOut", &gofbp.SubOut{}) // Basic SubOut + + net.Initialize("IN", proc1, "NAME") + net.Connect(proc1, "OUT", proc2, "IN", 6) + net.Connect(proc2, "COUNT", proc3, "IN", 6) + net.Initialize("OUT", proc3, "NAME") + + net.Run() +} diff --git a/subnets/subnet1.go b/subnets/subnet1.go new file mode 100644 index 0000000..df6ffb6 --- /dev/null +++ b/subnets/subnet1.go @@ -0,0 +1,27 @@ +package subnets + +import ( + "github.com/jpaulm/gofbp" + "github.com/jpaulm/gofbp/testrtn" +) + +type Subnet1 struct{} + +func (subnet *Subnet1) Setup(p *gofbp.Process) {} + +func (subnet *Subnet1) Execute(p *gofbp.Process) { + net := gofbp.NewSubnet("Subnet1", p) + + proc1 := net.NewProc("SubIn", &gofbp.SubIn{}) + + proc2 := net.NewProc("WriteToConsole1", &testrtn.WriteToConsole{}) + + proc3 := net.NewProc("SubOut", &gofbp.SubOut{}) + + net.Initialize("IN", proc1, "NAME") + net.Connect(proc1, "OUT", proc2, "IN", 6) + net.Connect(proc2, "OUT", proc3, "IN", 6) + net.Initialize("OUT", proc3, "NAME") + + net.Run() +} diff --git a/testing/force_deadlock_test.go b/testing/force_deadlock_test.go new file mode 100644 index 0000000..706f107 --- /dev/null +++ b/testing/force_deadlock_test.go @@ -0,0 +1,39 @@ +package testing + +import ( + "context" + "fmt" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestForceDeadlock(t *testing.T) { + exe := filepath.Join(t.TempDir(), "deadlock.exe") + + compile := exec.Command("go", "build", "-o", exe, "-v", "./testdata/deadlock") + data, err := compile.CombinedOutput() + s := string(data) + t.Log(s) + if err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Second) + defer cancel() + + run := exec.CommandContext(ctx, exe) + out, err := run.CombinedOutput() + s = string(out) + t.Log(s) + fmt.Println(s) + if !strings.Contains(s, "deadlock") && + !strings.Contains(s, "Deadlock") { + t.Error("expected output to mention a deadlock") + } + if err == nil { + t.Error("expected process to fail") + } +} diff --git a/gofbp_test.go b/testing/gofbp_test.go similarity index 99% rename from gofbp_test.go rename to testing/gofbp_test.go index 6c31142..5474be3 100644 --- a/gofbp_test.go +++ b/testing/gofbp_test.go @@ -1,4 +1,4 @@ -package main +package testing import ( "testing" diff --git a/testing/numbers.txt b/testing/numbers.txt new file mode 100644 index 0000000..e29b1db --- /dev/null +++ b/testing/numbers.txt @@ -0,0 +1,40 @@ +0 +1 +2 +3 +4 +5 +6 +7 +8 +9 +10 +11 +12 +13 +14 +15 +16 +17 +18 +19 +20 +21 +22 +23 +24 +25 +26 +27 +28 +29 +30 +31 +32 +33 +34 +35 +36 +37 +38 +39 diff --git a/testing/testdata/deadlock/main.go b/testing/testdata/deadlock/main.go new file mode 100644 index 0000000..2d8bb21 --- /dev/null +++ b/testing/testdata/deadlock/main.go @@ -0,0 +1,25 @@ +package main + +import ( + "github.com/jpaulm/gofbp/components/testrtn" + "github.com/jpaulm/gofbp/core" +) + +func main() { + net := core.NewNetwork("ForceDeadlock") + + proc1 := net.NewProc("Sender", &testrtn.Sender{}) + proc2 := net.NewProc("Counter", &testrtn.Counter{}) + proc3 := net.NewProc("Concat", &testrtn.ConcatStr{}) + + proc4 := net.NewProc("WriteToConsole", &testrtn.WriteToConsole{}) + + net.Initialize("15", proc1, "COUNT") + + net.Connect(proc1, "OUT", proc2, "IN", 6) + net.Connect(proc2, "OUT", proc3, "IN[1]", 6) + net.Connect(proc2, "COUNT", proc3, "IN[0]", 6) + net.Connect(proc3, "OUT", proc4, "IN", 6) + + net.Run() +} diff --git a/testing/testdata/infqueue b/testing/testdata/infqueue new file mode 100644 index 0000000..df23bfc --- /dev/null +++ b/testing/testdata/infqueue @@ -0,0 +1,40 @@ +IP - # 0 +IP - # 1 +IP - # 2 +IP - # 3 +IP - # 4 +IP - # 5 +IP - # 6 +IP - # 7 +IP - # 8 +IP - # 9 +IP - # 10 +IP - # 11 +IP - # 12 +IP - # 13 +IP - # 14 +IP - # 15 +IP - # 16 +IP - # 17 +IP - # 18 +IP - # 19 +IP - # 20 +IP - # 21 +IP - # 22 +IP - # 23 +IP - # 24 +IP - # 25 +IP - # 26 +IP - # 27 +IP - # 28 +IP - # 29 +IP - # 30 +IP - # 31 +IP - # 32 +IP - # 33 +IP - # 34 +IP - # 35 +IP - # 36 +IP - # 37 +IP - # 38 +IP - # 39 diff --git a/testing/testdata/testdata.txt b/testing/testdata/testdata.txt new file mode 100644 index 0000000..c7ff85e --- /dev/null +++ b/testing/testdata/testdata.txt @@ -0,0 +1,6 @@ +This repo holds the beginning of an FBP implementation in Go + +Features include: + +- delayed start of goroutines (FBP processes), unless `MustRun` attribute is specified or the process has no non-IIP inputs (same as JavaFBP delayed start feature) +- optional output ports - see https://github.com/jpaulm/gofbp/blob/master/components/testrtn/writetoconsole.go diff --git a/testinfqueueasmain_test.go b/testing/testinfqueueasmain_test.go similarity index 97% rename from testinfqueueasmain_test.go rename to testing/testinfqueueasmain_test.go index b52f20a..0cf9d37 100644 --- a/testinfqueueasmain_test.go +++ b/testing/testinfqueueasmain_test.go @@ -1,4 +1,4 @@ -package main +package testing import ( "testing" diff --git a/testing/testintsender_test.go b/testing/testintsender_test.go new file mode 100644 index 0000000..b3fe0d6 --- /dev/null +++ b/testing/testintsender_test.go @@ -0,0 +1,27 @@ +package testing + +import ( + "testing" + + "github.com/jpaulm/gofbp/components/io" + "github.com/jpaulm/gofbp/components/testrtn" + "github.com/jpaulm/gofbp/core" +) + +func TestIntSender(t *testing.T) { + params, err := core.LoadXMLParams("../params.xml") + if err != nil { + panic(err) + } + net := core.NewNetwork("TestIntSender") + net.SetParams(params) + proc1 := net.NewProc("IntSender", &testrtn.IntSender{}) + proc2 := net.NewProc("WriteToFile", &io.WriteFile{}) + proc3 := net.NewProc("WriteToConsole", &testrtn.WriteToConsole{}) + + net.Initialize("40", proc1, "COUNT") + net.Initialize("numbers.txt", proc2, "FILENAME") + net.Connect(proc1, "OUT", proc2, "IN", 6) + net.Connect(proc2, "OUT", proc3, "IN", 6) + net.Run() +} diff --git a/testing/testloadbal_test.go b/testing/testloadbal_test.go new file mode 100644 index 0000000..301aafc --- /dev/null +++ b/testing/testloadbal_test.go @@ -0,0 +1,28 @@ +package testing + +import ( + "testing" + + "github.com/jpaulm/gofbp/components/testrtn" + "github.com/jpaulm/gofbp/core" +) + +func TestLoadBal(t *testing.T) { + net := core.NewNetwork("TestLoadBal") + + sender := net.NewProc("Sender", &testrtn.Sender{}) + + proc2 := net.NewProc("LoadBalance", &testrtn.LoadBalance{}) + + proc3a := net.NewProc("Receiver0", &testrtn.Receiver{}) + proc3b := net.NewProc("Receiver1", &testrtn.DelayedReceiver{}) + proc3c := net.NewProc("Receiver2", &testrtn.DelayedReceiver{}) + + net.Initialize("40", sender, "COUNT") + net.Connect(sender, "OUT", proc2, "IN", 6) + net.Connect(proc2, "OUT[0]", proc3a, "IN", 6) + net.Connect(proc2, "OUT[1]", proc3b, "IN", 6) + net.Connect(proc2, "OUT[2]", proc3c, "IN", 6) + + net.Run() +} diff --git a/testsubnet1_test.go b/testing/testsubnet1_test.go similarity index 97% rename from testsubnet1_test.go rename to testing/testsubnet1_test.go index a97d83b..fad21e3 100644 --- a/testsubnet1_test.go +++ b/testing/testsubnet1_test.go @@ -1,4 +1,4 @@ -package main +package testing import ( "testing" diff --git a/testsubnet2_test.go b/testing/testsubnet2_test.go similarity index 97% rename from testsubnet2_test.go rename to testing/testsubnet2_test.go index daa839d..c72a67f 100644 --- a/testsubnet2_test.go +++ b/testing/testsubnet2_test.go @@ -1,4 +1,4 @@ -package main +package testing // Thiis differs from TestSubnet1 because it uses SSSubnet1 (SubstreamSenstive) instead of simple Subnet1... diff --git a/testsubnet3_test.go b/testing/testsubnet3_test.go similarity index 98% rename from testsubnet3_test.go rename to testing/testsubnet3_test.go index d8f0637..2b2e1c1 100644 --- a/testsubnet3_test.go +++ b/testing/testsubnet3_test.go @@ -1,4 +1,4 @@ -package main +package testing // Thiis differs from TestSubnet1 because it uses SSSubnet1 (SubstreamSenstive) instead of simple Subnet1... diff --git a/testing/testwebsocket_test.go b/testing/testwebsocket_test.go new file mode 100644 index 0000000..7ba5bca --- /dev/null +++ b/testing/testwebsocket_test.go @@ -0,0 +1,27 @@ +package testing + +import ( + "testing" + + "github.com/jpaulm/gofbp/components/websocket" + "github.com/jpaulm/gofbp/components/websocket/ans" + "github.com/jpaulm/gofbp/core" +) + +func TestWebSocket(t *testing.T) { + params, err := core.LoadXMLParams("../params.xml") + if err != nil { + panic(err) + } + path := "localhost:8080" + + net := core.NewNetwork("TestWebSocket") + net.SetParams(params) + proc1 := net.NewProc("WSRequest", &websocket.WSRequest{}) + proc2 := net.NewProc("AnsReq", &ans.WSAnsReq{}) + proc3 := net.NewProc("WSRespond", &websocket.WSRespond{}) + net.Initialize(path, proc1, "ADDR") + net.Connect(proc1, "OUT", proc2, "IN", 6) + net.Connect(proc2, "OUT", proc3, "IN", 6) + net.Run() +} diff --git a/testrtn/concatstr.go b/testrtn/concatstr.go new file mode 100644 index 0000000..5857e6b --- /dev/null +++ b/testrtn/concatstr.go @@ -0,0 +1,33 @@ +package testrtn + +import "github.com/jpaulm/gofbp" + +type ConcatStr struct { + ipt gofbp.InputArrayConn + opt gofbp.OutputConn +} + +func (concatstr *ConcatStr) Setup(p *gofbp.Process) { + concatstr.ipt = p.OpenInArrayPort("IN") + concatstr.opt = p.OpenOutPort("OUT") +} + +func (concatstr *ConcatStr) Execute(p *gofbp.Process) { + + for i := 0; i < concatstr.ipt.ArrayLength(); i++ { + + for { + conn := concatstr.ipt.GetArrayItem(i) + if conn == nil { + continue + } + var pkt = p.Receive(conn) + if pkt == nil { + break + } + //fmt.Println("Output: ", pkt.Contents) + p.Send(concatstr.opt, pkt) + } + } + +} diff --git a/testrtn/counter.go b/testrtn/counter.go new file mode 100644 index 0000000..81858f7 --- /dev/null +++ b/testrtn/counter.go @@ -0,0 +1,45 @@ +package testrtn + +import ( + //"fmt" + "strconv" + + "github.com/jpaulm/gofbp" +) + +type Counter struct { + ipt gofbp.InputConn + cnt gofbp.OutputConn + opt gofbp.OutputConn +} + +func (counter *Counter) Setup(p *gofbp.Process) { + counter.ipt = p.OpenInPort("IN") + counter.cnt = p.OpenOutPort("COUNT") + counter.opt = p.OpenOutPortOptional("OUT") +} + +func (Counter) MustRun() {} + +func (counter *Counter) Execute(p *gofbp.Process) { + + count := 0 + + for { + var pkt = p.Receive(counter.ipt) + if pkt == nil { + break + } + if counter.opt.IsConnected() { + p.Send(counter.opt, pkt) + } else { + p.Discard(pkt) + } + + count++ + } + + pkt := p.Create(strconv.Itoa(count)) + p.Send(counter.cnt, pkt) + +} diff --git a/testrtn/delayedreceiver.go b/testrtn/delayedreceiver.go new file mode 100644 index 0000000..fda03f1 --- /dev/null +++ b/testrtn/delayedreceiver.go @@ -0,0 +1,32 @@ +package testrtn + +import ( + "fmt" + "time" + + "github.com/jpaulm/gofbp" +) + +type DelayedReceiver struct { + ipt gofbp.InputConn +} + +func (receiver *DelayedReceiver) Setup(p *gofbp.Process) { + receiver.ipt = p.OpenInPort("IN") +} + +func (DelayedReceiver) MustRun() {} + +func (receiver *DelayedReceiver) Execute(p *gofbp.Process) { + + for { + var pkt = p.Receive(receiver.ipt) + if pkt == nil { + break + } + time.Sleep(500 * time.Millisecond) + fmt.Println("Input to DelayedReceiver:", p.Name, ">", pkt.Contents) + p.Discard(pkt) + } + +} diff --git a/testrtn/discard.go b/testrtn/discard.go new file mode 100644 index 0000000..832cde9 --- /dev/null +++ b/testrtn/discard.go @@ -0,0 +1,32 @@ +package testrtn + +import ( + //"fmt" + + "github.com/jpaulm/gofbp" +) + +type Discard struct { + ipt gofbp.InputConn +} + +func (discard *Discard) Setup(p *gofbp.Process) { + discard.ipt = p.OpenInPort("IN") +} + +//func (Discard) MustRun() {} + +func (discard *Discard) Execute(p *gofbp.Process) { + + for { + var pkt = p.Receive(discard.ipt) + if pkt == nil { + break + } + //fmt.Println(pkt.Contents) + + p.Discard(pkt) + + } + +} diff --git a/testrtn/kick.go b/testrtn/kick.go new file mode 100644 index 0000000..7fe58ba --- /dev/null +++ b/testrtn/kick.go @@ -0,0 +1,22 @@ +package testrtn + +import ( + //"fmt" + + "github.com/jpaulm/gofbp" +) + +type Kick struct { + opt gofbp.OutputConn +} + +func (kick *Kick) Setup(p *gofbp.Process) { + kick.opt = p.OpenOutPort("OUT") +} + +func (kick *Kick) Execute(p *gofbp.Process) { + + var pkt = p.Create("Kicker IP") + p.Send(kick.opt, pkt) + +} diff --git a/testrtn/loadbalance.go b/testrtn/loadbalance.go new file mode 100644 index 0000000..7f91c5a --- /dev/null +++ b/testrtn/loadbalance.go @@ -0,0 +1,47 @@ +package testrtn + +import ( + "github.com/jpaulm/gofbp" +) + +type LoadBalance struct { + ipt gofbp.InputConn + out gofbp.OutputArrayConn +} + +func (loadbalance *LoadBalance) Setup(p *gofbp.Process) { + loadbalance.ipt = p.OpenInPort("IN") + loadbalance.out = p.OpenOutArrayPort("OUT") +} + +func (loadbalance *LoadBalance) Execute(p *gofbp.Process) { + + var i int + var level int + var inSubstream bool + for { + var pkt = p.Receive(loadbalance.ipt) + if pkt == nil { + break + } + + if !inSubstream { + i = loadbalance.out.GetItemWithFewestIPs() + } + + if pkt.PktType == gofbp.OpenBracket { + inSubstream = true + level++ + } + + if pkt.PktType == gofbp.CloseBracket && level == 1 { + inSubstream = false + level-- + } + + opt := loadbalance.out.GetArrayItem(i) + + p.Send(opt, pkt) + } + +} diff --git a/testrtn/prefix.go b/testrtn/prefix.go new file mode 100644 index 0000000..a8374d5 --- /dev/null +++ b/testrtn/prefix.go @@ -0,0 +1,49 @@ +package testrtn + +import ( + "fmt" + + "github.com/jpaulm/gofbp" +) + +type Prefix struct { + ipt gofbp.InputConn + iptIP gofbp.InputConn + out gofbp.OutputConn +} + +func (prefix *Prefix) Setup(p *gofbp.Process) { + + prefix.ipt = p.OpenInPort("IN") + + prefix.out = p.OpenOutPort("OUT") + + prefix.iptIP = p.OpenInPort("PARAM") +} + +func (prefix *Prefix) Execute(p *gofbp.Process) { + + icpkt := p.Receive(prefix.iptIP) + param := icpkt.Contents.(string) + + p.Discard(icpkt) + + p.Close(prefix.iptIP) + + for { + var pkt = p.Receive(prefix.ipt) + if pkt == nil { + break + } + fmt.Println(pkt.Contents) + + s, ok := pkt.Contents.(string) + if !ok { + panic("IP contents must be a string") + } + s = param + s + pkt.Contents = s + p.Send(prefix.out, pkt) + + } +} diff --git a/testrtn/receiver.go b/testrtn/receiver.go new file mode 100644 index 0000000..19c4549 --- /dev/null +++ b/testrtn/receiver.go @@ -0,0 +1,31 @@ +package testrtn + +import ( + "fmt" + + "github.com/jpaulm/gofbp" +) + +type Receiver struct { + ipt gofbp.InputConn +} + +func (receiver *Receiver) Setup(p *gofbp.Process) { + receiver.ipt = p.OpenInPort("IN") +} + +func (Receiver) MustRun() {} + +func (receiver *Receiver) Execute(p *gofbp.Process) { + + for { + var pkt = p.Receive(receiver.ipt) + if pkt == nil { + break + } + + fmt.Println("Input to Receiver:", p.Name, ">", pkt.Contents) + p.Discard(pkt) + } + +} diff --git a/testrtn/roundrobinsender.go b/testrtn/roundrobinsender.go new file mode 100644 index 0000000..fd499ca --- /dev/null +++ b/testrtn/roundrobinsender.go @@ -0,0 +1,36 @@ +package testrtn + +import ( + "github.com/jpaulm/gofbp" +) + +type RoundRobinSender struct { + ipt gofbp.InputConn + out gofbp.OutputArrayConn +} + +func (rrsender *RoundRobinSender) Setup(p *gofbp.Process) { + rrsender.ipt = p.OpenInPort("IN") + rrsender.out = p.OpenOutArrayPort("OUT") +} + +func (rrsender *RoundRobinSender) Execute(p *gofbp.Process) { + + var i = 0 + + j := rrsender.out.ArrayLength() + + for { + var pkt = p.Receive(rrsender.ipt) + if pkt == nil { + break + } + //fmt.Println("Output: ", pkt.Contents) + + opt := rrsender.out.GetArrayItem(i) + + p.Send(opt, pkt) + i = (i + 1) % j + } + +} diff --git a/testrtn/selector.go b/testrtn/selector.go new file mode 100644 index 0000000..29be0a6 --- /dev/null +++ b/testrtn/selector.go @@ -0,0 +1,62 @@ +package testrtn + +import ( + "fmt" + "strings" + + "github.com/jpaulm/gofbp" +) + +type Selector struct { + ipt gofbp.InputConn + iptIP gofbp.InputConn + out1 gofbp.OutputConn + out2 gofbp.OutputConn +} + +func (selector *Selector) Setup(p *gofbp.Process) { + + selector.ipt = p.OpenInPort("IN") + + selector.out1 = p.OpenOutPort("ACC") + + selector.out2 = p.OpenOutPortOptional("REJ") + + selector.iptIP = p.OpenInPort("PARAM") +} + +func (Selector) MustRun() {} + +func (selector *Selector) Execute(p *gofbp.Process) { + + icpkt := p.Receive(selector.iptIP) + param := icpkt.Contents.(string) + i := len(param) + + p.Discard(icpkt) + + p.Close(selector.iptIP) + + for { + var pkt = p.Receive(selector.ipt) + if pkt == nil { + break + } + fmt.Println(pkt.Contents) + + s := pkt.Contents.(string) + if i > len(s) { + i = len(s) + } + + if 0 == strings.Compare(param[:i], s[:i]) { + p.Send(selector.out1, pkt) + } else { + if !selector.out2.IsConnected() { + p.Discard(pkt) + } else { + p.Send(selector.out2, pkt) + } + } + } +} diff --git a/testrtn/sender.go b/testrtn/sender.go new file mode 100644 index 0000000..bab580c --- /dev/null +++ b/testrtn/sender.go @@ -0,0 +1,33 @@ +package testrtn + +import ( + //"fmt" + "strconv" + + "github.com/jpaulm/gofbp" +) + +type Sender struct { + ipt gofbp.InputConn + opt gofbp.OutputConn +} + +func (sender *Sender) Setup(p *gofbp.Process) { + sender.ipt = p.OpenInPort("COUNT") + sender.opt = p.OpenOutPort("OUT") +} + +func (sender *Sender) Execute(p *gofbp.Process) { + + icpkt := p.Receive(sender.ipt) + j, _ := strconv.Atoi(icpkt.Contents.(string)) + p.Discard(icpkt) + p.Close(sender.ipt) + + var pkt *gofbp.Packet + for i := 0; i < j; i++ { + pkt = p.Create("IP - # " + strconv.Itoa(i)) + p.Send(sender.opt, pkt) + } + +} diff --git a/testrtn/substreamsender.go b/testrtn/substreamsender.go new file mode 100644 index 0000000..c98aa93 --- /dev/null +++ b/testrtn/substreamsender.go @@ -0,0 +1,55 @@ +package testrtn + +import ( + "strconv" + + "github.com/jpaulm/gofbp" +) + +type SubstreamSender struct { + ipt gofbp.InputConn + opt gofbp.OutputConn +} + +func (sender *SubstreamSender) Setup(p *gofbp.Process) { + sender.ipt = p.OpenInPort("COUNT") + sender.opt = p.OpenOutPort("OUT") +} + +func (sender *SubstreamSender) Execute(p *gofbp.Process) { + + icpkt := p.Receive(sender.ipt) + j, _ := strconv.Atoi(icpkt.Contents.(string)) + p.Discard(icpkt) + p.Close(sender.ipt) + + var pkt *gofbp.Packet + pkt = p.CreateBracket(gofbp.OpenBracket, "") + p.Send(sender.opt, pkt) + + for i := 0; i < j; i++ { + k := i % 10 + + if k == 2 { + pkt = p.CreateBracket(gofbp.CloseBracket, "") + p.Send(sender.opt, pkt) + } + if k == 3 { + pkt = p.CreateBracket(gofbp.OpenBracket, "") + p.Send(sender.opt, pkt) + } + + if k == 7 || k == 0 { + pkt = p.CreateBracket(gofbp.CloseBracket, "") + p.Send(sender.opt, pkt) + pkt = p.CreateBracket(gofbp.OpenBracket, "") + p.Send(sender.opt, pkt) + } + + pkt = p.Create("IP - # " + strconv.Itoa(i)) + p.Send(sender.opt, pkt) + } + pkt = p.CreateBracket(gofbp.CloseBracket, "") + p.Send(sender.opt, pkt) + +} diff --git a/testrtn/writetoconsnl.go b/testrtn/writetoconsnl.go new file mode 100644 index 0000000..abf522a --- /dev/null +++ b/testrtn/writetoconsnl.go @@ -0,0 +1,48 @@ +package testrtn + +import ( + "fmt" + + "github.com/jpaulm/gofbp" +) + +// WriteToConsole modified to be a non-looper (NL) + +type WriteToConsNL struct { + ipt gofbp.InputConn + opt gofbp.OutputConn +} + +func (writeToConsole *WriteToConsNL) Setup(p *gofbp.Process) { + writeToConsole.ipt = p.OpenInPort("IN") + writeToConsole.opt = p.OpenOutPortOptional("OUT") +} + +//func (WriteToConsNL) MustRun() {} + +func (writeToConsole *WriteToConsNL) Execute(p *gofbp.Process) { + + //for { + var pkt = p.Receive(writeToConsole.ipt) + if pkt == nil { + //break + return + } + if pkt.PktType == gofbp.OpenBracket { + fmt.Println("Open", pkt.Contents) + } else { + if pkt.PktType == gofbp.CloseBracket { + fmt.Println("Close", pkt.Contents) + } else { + fmt.Println(pkt.Contents) + } + } + + if writeToConsole.opt.IsConnected() { + p.Send(writeToConsole.opt, pkt) + } else { + p.Discard(pkt) + } + //} + +} diff --git a/testrtn/writetoconsole.go b/testrtn/writetoconsole.go new file mode 100644 index 0000000..7626e93 --- /dev/null +++ b/testrtn/writetoconsole.go @@ -0,0 +1,45 @@ +package testrtn + +import ( + "fmt" + + "github.com/jpaulm/gofbp" +) + +type WriteToConsole struct { + ipt gofbp.InputConn + out gofbp.OutputConn +} + +func (writeToConsole *WriteToConsole) Setup(p *gofbp.Process) { + writeToConsole.ipt = p.OpenInPort("IN") + writeToConsole.out = p.OpenOutPortOptional("OUT") +} + +func (WriteToConsole) MustRun() {} + +func (writeToConsole *WriteToConsole) Execute(p *gofbp.Process) { + + for { + var pkt = p.Receive(writeToConsole.ipt) + if pkt == nil { + break + } + if pkt.PktType == gofbp.OpenBracket { + fmt.Println("Open", pkt.Contents) + } else { + if pkt.PktType == gofbp.CloseBracket { + fmt.Println("Close", pkt.Contents) + } else { + fmt.Println(pkt.Contents) + } + } + + if writeToConsole.out.IsConnected() { + p.Send(writeToConsole.out, pkt) + } else { + p.Discard(pkt) + } + } + +}