lourenco/mgmt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

lang: Add module imports and more

Browse Source 
This enables imports in mcl code, and is one of last remaining blockers
to using mgmt. Now we can start writing standalone modules, and adding
standard library functions as needed. There's still lots to do, but this
was a big missing piece. It was much harder to get right than I had
expected, but I think it's solid!

This unfortunately large commit is the result of some wild hacking I've
been doing for the past little while. It's the result of a rebase that
broke many "wip" commits that tracked my private progress, into
something that's not gratuitously messy for our git logs. Since this was
a learning and discovery process for me, I've "erased" the confusing git
history that wouldn't have helped. I'm happy to discuss the dead-ends,
and a small portion of that code was even left in for possible future
use.

This patch includes:

* A change to the cli interface:
You now specify the front-end explicitly, instead of leaving it up to
the front-end to decide when to "activate". For example, instead of:

mgmt run --lang code.mcl

we now do:

mgmt run lang --lang code.mcl

We might rename the --lang flag in the future to avoid the awkward word
repetition. Suggestions welcome, but I'm considering "input". One
side-effect of this change, is that flags which are "engine" specific
now must be specified with "run" before the front-end name. Eg:

mgmt run --tmp-prefix lang --lang code.mcl

instead of putting --tmp-prefix at the end. We also changed the GAPI
slightly, but I've patched all code that used it. This also makes things
consistent with the "deploy" command.

* The deploys are more robust and let you deploy after a run
This has been vastly improved and let's mgmt really run as a smart
engine that can handle different workloads. If you don't want to deploy
when you've started with `run` or if one comes in, you can use the
--no-watch-deploy option to block new deploys.

* The import statement exists and works!
We now have a working `import` statement. Read the docs, and try it out.
I think it's quite elegant how it fits in with `SetScope`. Have a look.
As a result, we now have some built-in functions available in modules.
This also adds the metadata.yaml entry-point for all modules. Have a
look at the examples or the tests. The bulk of the patch is to support
this.

* Improved lang input parsing code:
I re-wrote the parsing that determined what ran when we passed different
things to --lang. Deciding between running an mcl file or raw code is
now handled in a more intelligent, and re-usable way. See the inputs.go
file if you want to have a look. One casualty is that you can't stream
code from stdin *directly* to the front-end, it's encapsulated into a
deploy first. You can still use stdin though! I doubt anyone will notice
this change.

* The scope was extended to include functions and classes:
Go forth and import lovely code. All these exist in scopes now, and can
be re-used!

* Function calls actually use the scope now. Glad I got this sorted out.

* There is import cycle detection for modules!
Yes, this is another dag. I think that's #4. I guess they're useful.

* A ton of tests and new test infra was added!
This should make it much easier to add new tests that run mcl code. Have
a look at TestAstFunc1 to see how to add more of these.

As usual, I'll try to keep these commits smaller in the future!
This commit is contained in:
James Shubin
2018-11-22 16:48:10 -05:00
parent 948a3c6d08
commit 96dccca475
146 changed files with 5301 additions and 1112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

209
lib/main.go
View File

@@ -23,6 +23,7 @@ import (
"log"
"os"
"path"
"strings"
"sync"
"time"
@@ -70,6 +71,7 @@ type Main struct {
NoWatch bool // do not change graph under any circumstances
NoConfigWatch bool // do not update graph due to config changes
NoStreamWatch bool // do not update graph due to stream changes
NoDeployWatch bool // do not change deploys after an initial deploy
Noop bool // globally force all resources into no-op mode
Sema int // add a semaphore with this lock count to each resource
@@ -114,6 +116,9 @@ func (obj *Main) Validate() error {
if obj.Program == "" || obj.Version == "" {
return fmt.Errorf("you must set the Program and Version strings")
}
if strings.Contains(obj.Program, " ") {
return fmt.Errorf("the Program string contains unexpected spaces")
}
if obj.Prefix != nil && obj.TmpPrefix {
return fmt.Errorf("choosing a prefix and the request for a tmp prefix is illogical")
@@ -139,7 +144,7 @@ func (obj *Main) Init() error {
}
if obj.idealClusterSize < 1 {
return fmt.Errorf("the IdealClusterSize should be at least one")
return fmt.Errorf("the IdealClusterSize (%d) should be at least one", obj.idealClusterSize)
}
// transform the url list inputs into etcd typed lists
@@ -187,7 +192,7 @@ func (obj *Main) Run() error {
}
hello(obj.Program, obj.Version, obj.Flags) // say hello!
defer Logf("Goodbye!")
defer Logf("goodbye!")
defer obj.exit.Done(nil) // ensure this gets called even if Exit doesn't
@@ -216,7 +221,7 @@ func (obj *Main) Run() error {
Logf("warning: working prefix directory is temporary!")
} else {
return fmt.Errorf("can't create prefix")
return fmt.Errorf("can't create prefix: `%s`", prefix)
}
}
Logf("working prefix is: %s", prefix)
@@ -472,7 +477,7 @@ func (obj *Main) Run() error {
}
gapiImpl = gapiObj // copy it to active
data := gapi.Data{
data := &gapi.Data{
Program: obj.Program,
Hostname: hostname,
World: world,
@@ -666,109 +671,151 @@ func (obj *Main) Run() error {
}
}()
if obj.Deploy != nil {
deploy := obj.Deploy
// redundant
deploy.Noop = obj.Noop
deploy.Sema = obj.Sema
// get max id (from all the previous deploys)
// this is what the existing cluster is already running
// TODO: can this block since we didn't deploy yet?
max, err := etcd.GetMaxDeployID(embdEtcd)
if err != nil {
return errwrap.Wrapf(err, "error getting max deploy id")
}
select {
case deployChan <- deploy:
// send
case <-exitchan:
// pass
}
// improved etcd based deploy
wg.Add(1)
go func() {
defer wg.Done()
defer close(deployChan) // no more are coming ever!
// don't inline this, because when we close the deployChan it's
// the signal to tell the engine to actually shutdown...
wg.Add(1)
go func() {
defer wg.Done()
defer close(deployChan) // no more are coming ever!
select { // wait until we're ready to shutdown
// we've been asked to deploy, so do that first...
if obj.Deploy != nil {
deploy := obj.Deploy
// redundant
deploy.Noop = obj.Noop
deploy.Sema = obj.Sema
select {
case deployChan <- deploy:
// send
if obj.Flags.Debug {
Logf("deploy: sending new gapi")
}
case <-exitchan:
return
}
}()
} else {
// etcd based deploy
wg.Add(1)
go func() {
defer wg.Done()
defer close(deployChan)
startChan := make(chan struct{}) // start signal
close(startChan) // kick it off!
for {
select {
case <-startChan: // kick the loop once at start
startChan = nil // disable
case err, ok := <-etcd.WatchDeploy(embdEtcd):
if !ok {
obj.exit.Done(nil) // regular shutdown
return
}
if err != nil {
// TODO: it broke, can we restart?
obj.exit.Done(fmt.Errorf("deploy: watch error"))
return
}
startChan = nil // disable it early...
}
// now we can wait for future deploys, but if we already had an
// initial deploy from run, don't switch to this unless it's new
var last uint64
startChan := make(chan struct{}) // start signal
close(startChan) // kick it off!
for {
if obj.NoDeployWatch && (obj.Deploy != nil || last > 0) {
// block here, because when we close the
// deployChan it's the signal to tell the engine
// to actually shutdown...
select { // wait until we're ready to shutdown
case <-exitchan:
return
}
}
select {
case <-startChan: // kick the loop once at start
startChan = nil // disable
case err, ok := <-etcd.WatchDeploy(embdEtcd):
if !ok {
obj.exit.Done(nil) // regular shutdown
return
}
if err != nil {
// TODO: it broke, can we restart?
obj.exit.Done(fmt.Errorf("deploy: watch error"))
return
}
startChan = nil // disable it early...
if obj.Flags.Debug {
Logf("deploy: got activity")
}
str, err := etcd.GetDeploy(embdEtcd, 0) // 0 means get the latest one
if err != nil {
Logf("deploy: error getting deploy: %+v", err)
continue
}
if str == "" { // no available deploys exist yet
// send an empty deploy... this is done
// to start up the engine so it can run
// an empty graph and be ready to swap!
Logf("deploy: empty")
deploy := &gapi.Deploy{
Name: empty.Name,
GAPI: &empty.GAPI{},
}
select {
case deployChan <- deploy:
// send
if obj.Flags.Debug {
Logf("deploy: sending empty deploy")
}
case <-exitchan:
return
}
continue
}
case <-exitchan:
return
}
// decode the deploy (incl. GAPI) and send it!
deploy, err := gapi.NewDeployFromB64(str)
if err != nil {
Logf("deploy: error decoding deploy: %+v", err)
continue
}
latest, err := etcd.GetMaxDeployID(embdEtcd) // or zero
if err != nil {
Logf("error getting max deploy id: %+v", err)
continue
}
// if we already did the built-in one from run, and this
// new deploy is not newer than when we started, skip it
if obj.Deploy != nil && latest <= max {
// if latest and max are zero, it's okay to loop
continue
}
// if we're doing any deploy, don't run the previous one
// (this might be useful if we get a double event here!)
if obj.Deploy == nil && latest <= last && latest != 0 {
// if latest and last are zero, pass through it!
continue
}
// if we already did a deploy, but we're being asked for
// this again, then skip over it if it's not a newer one
if obj.Deploy != nil && latest <= last {
continue
}
// 0 passes through an empty deploy without an error...
// (unless there is some sort of etcd error that occurs)
str, err := etcd.GetDeploy(embdEtcd, latest)
if err != nil {
Logf("deploy: error getting deploy: %+v", err)
continue
}
if str == "" { // no available deploys exist yet
// send an empty deploy... this is done
// to start up the engine so it can run
// an empty graph and be ready to swap!
Logf("deploy: empty")
deploy := &gapi.Deploy{
Name: empty.Name,
GAPI: &empty.GAPI{},
}
select {
case deployChan <- deploy:
// send
if obj.Flags.Debug {
Logf("deploy: sending new gapi")
Logf("deploy: sending empty deploy")
}
case <-exitchan:
return
}
continue
}
}()
}
// decode the deploy (incl. GAPI) and send it!
deploy, err := gapi.NewDeployFromB64(str)
if err != nil {
Logf("deploy: error decoding deploy: %+v", err)
continue
}
select {
case deployChan <- deploy:
last = latest // update last deployed
// send
if obj.Flags.Debug {
Logf("deploy: sent new gapi")
}
case <-exitchan:
return
}
}
}()
Logf("running...")