Commit 5dd6dd57 authored by Kirill Smelkov's avatar Kirill Smelkov

.

parent 6410356a
...@@ -181,7 +181,7 @@ func (c *Client) updateOperational() (sendReady func()) { ...@@ -181,7 +181,7 @@ func (c *Client) updateOperational() (sendReady func()) {
// If successful it returns with operational state RLocked (c.node.StateMu) and // If successful it returns with operational state RLocked (c.node.StateMu) and
// unlocked otherwise. // unlocked otherwise.
// //
// The only error possible is if provided ctx cancel. // The only error possible is if provided ctx cancels.
// XXX and client stopped/closed? (ctx passed to Run cancelled) // XXX and client stopped/closed? (ctx passed to Run cancelled)
func (c *Client) withOperational(ctx context.Context) error { func (c *Client) withOperational(ctx context.Context) error {
for { for {
...@@ -259,7 +259,7 @@ func (c *Client) talkMaster1(ctx context.Context) (err error) { ...@@ -259,7 +259,7 @@ func (c *Client) talkMaster1(ctx context.Context) (err error) {
c.mlinkMu.Unlock() c.mlinkMu.Unlock()
close(ready) close(ready)
wg, ctx := errgroup.WithContext(ctx) wg, ctx := errgroup.WithContext(ctx) // XXX -> xsync.WorkGroup
defer xio.CloseWhenDone(ctx, mlink)() defer xio.CloseWhenDone(ctx, mlink)()
......
// Copyright (C) 2020 Nexedi SA and Contributors.
// Kirill Smelkov <kirr@nexedi.com>
//
// This program is free software: you can Use, Study, Modify and Redistribute
// it under the terms of the GNU General Public License version 3, or (at your
// option) any later version, as published by the Free Software Foundation.
//
// You can also Link and Combine this program with other software covered by
// the terms of any of the Free Software licenses or any of the Open Source
// Initiative approved licenses and Convey the resulting work. Corresponding
// source of such a combination shall include the source code for all other
// software used.
//
// This program is distributed WITHOUT ANY WARRANTY; without even the implied
// warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
//
// See COPYING file for full licensing terms.
// See https://www.nexedi.com/licensing for rationale and options.
package neo
import (
"context"
"io/ioutil"
"os"
"os/exec"
"time"
"lab.nexedi.com/kirr/go123/xerr"
)
// NEOSrv represents running NEO server.
type NEOSrv interface {
MasterAddr() string // address of the master
// XXX +ClusterName
}
// NEOPySrv represents running NEO/py server.
//
// Create it with StartNEOPySrv(XXX).
type NEOPySrv struct {
pysrv *exec.Cmd // spawned `XXX`
workdir string // location for database and log files
opt NEOPyOptions // options for spawned server
cancel func() // to stop pysrv
done chan struct{} // ready after Wait completes
errExit error // error from Wait
masterAddr string // address of master in spawned cluster
}
// NEOPySrv.Bugs
type NEOPyOptions struct {
// nmaster
// npartition
// nreplica
// name
}
// StartNEOPySrv starts NEO/py server for NEO database located in workdir/.
func StartNEOPySrv(workdir string, opt NEOPyOptions) (_ *NEOPySrv, err error) {
defer xerr.Contextf(&err, "startneo %s", workdir)
ctx, cancel := context.WithCancel(context.Background())
readyf := workdir + "/ready"
err = os.Remove(readyf)
if os.IsNotExist(err) {
err = nil
}
if err != nil {
return nil, err
}
n := &NEOPySrv{workdir: workdir, cancel: cancel, done: make(chan struct{})}
n.pysrv = exec.CommandContext(ctx, "python", "./py/runneo.py", workdir) // XXX +opt
n.opt = opt
n.pysrv.Stdin = nil
n.pysrv.Stdout = os.Stdout
n.pysrv.Stderr = os.Stderr
err = n.pysrv.Start()
if err != nil {
return nil, err
}
go func() {
n.errExit = n.pysrv.Wait()
close(n.done)
}()
defer func() {
if err != nil {
n.Close()
}
}()
// wait till spawned NEO is ready to serve clients
for {
select {
default:
case <-n.done:
return nil, n.errExit
}
_, err := os.Stat(readyf)
if err == nil {
break // NEO cluster spawned by runneo.py is running
}
if os.IsNotExist(err) {
err = nil // not yet
}
if err != nil {
return nil, err
}
time.Sleep(100*time.Millisecond)
}
// retrieve master address
masterAddr, err := ioutil.ReadFile(readyf)
if err != nil {
return nil, err
}
n.masterAddr = string(masterAddr)
return n, nil
}
func (n *NEOPySrv) MasterAddr() string {
return n.masterAddr
}
func (n *NEOPySrv) Close() (err error) {
defer xerr.Contextf(&err, "stopneo %s", n.workdir)
n.cancel()
<-n.done
err = n.errExit
if _, ok := err.(*exec.ExitError); ok {
err = nil // ignore exit status - it is always !0 on kill
}
return err
}
# -*- coding: utf-8 -*-
# Copyright (C) 2020 Nexedi SA and Contributors.
# Kirill Smelkov <kirr@nexedi.com>
#
# This program is free software: you can Use, Study, Modify and Redistribute
# it under the terms of the GNU General Public License version 3, or (at your
# option) any later version, as published by the Free Software Foundation.
#
# You can also Link and Combine this program with other software covered by
# the terms of any of the Free Software licenses or any of the Open Source
# Initiative approved licenses and Convey the resulting work. Corresponding
# source of such a combination shall include the source code for all other
# software used.
#
# This program is distributed WITHOUT ANY WARRANTY; without even the implied
# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
#
# See COPYING file for full licensing terms.
# See https://www.nexedi.com/licensing for rationale and options.
"""runneo.py runs NEO/py cluster for NEO/go testing.
Usage: runneopy <workdir> <readyfile> XXX + (**kw for NEOCluster)
XXX
"""
from neo.tests.functional import NEOCluster
from golang import func, defer
@func
def main():
workdir = sys.argv[1]
readyf = sys.argv[1]
cluster = NEOCluster(['1'], adapter='SQLite', temp_dir=workdir) # XXX +kw
cluster.start()
defer(cluster.stop)
cluster.expectClusterRunning()
zstor = cluster.getZODBStorage()
# dump information about ready cluster into readyfile
with open("%s.tmp" % readyf, "w") as f:
# XXX master addresses
# XXX + addresses of other nodes?
f.write("...")
os.rename("%s.tmp" % readyf, readyf) # atomic
def _():
os.unlink(readyf)
defer(_)
# XXX loop forever
if __name__ == '__main__':
main()
...@@ -138,7 +138,7 @@ func (z *ZEOPySrv) Close() (err error) { ...@@ -138,7 +138,7 @@ func (z *ZEOPySrv) Close() (err error) {
<-z.done <-z.done
err = z.errExit err = z.errExit
if _, ok := err.(*exec.ExitError); ok { if _, ok := err.(*exec.ExitError); ok {
err = nil // ignore exit statue - it is always !0 on kill err = nil // ignore exit status - it is always !0 on kill
} }
return err return err
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment