Commit 95d0183c authored by Kirill Smelkov's avatar Kirill Smelkov

.

parent d77ccb51
......@@ -136,7 +136,7 @@ func (nt *NodeTable) GetByLink(link *NodeLink) *Node {
}
// XXX doc
func (nt *NodeTable) SetNodeState(node *Node, state neo.NodeState) {
func (nt *NodeTable) SetNodeState(node *Node, state NodeState) {
node.NodeState = state
nt.notify(node.NodeInfo)
}
......
This diff is collapsed.
......@@ -79,7 +79,8 @@ func Serve(ctx context.Context, l *neo.Listener, srv Server) error {
}
*/
/*
// FIXME kill vvv
///*
// ----------------------------------------
// XXX goes away? (we need a func to make sure to recv RequestIdentification
......@@ -129,4 +130,4 @@ func IdentifyPeer(link *neo.NodeLink, myNodeType neo.NodeType) (nodeInfo neo.Req
return req, nil
}
*/
//*/
......@@ -28,6 +28,7 @@ import (
"lab.nexedi.com/kirr/neo/go/neo"
"lab.nexedi.com/kirr/neo/go/zodb"
"lab.nexedi.com/kirr/neo/go/xcommon/log"
"lab.nexedi.com/kirr/neo/go/xcommon/xnet"
"lab.nexedi.com/kirr/neo/go/xcommon/xcontext"
......@@ -92,7 +93,7 @@ func (stor *Storage) Run(ctx context.Context) error {
if err != nil {
return err // XXX err ctx
}
stor.logf("serving on %s ...", l.Addr())
log.Infof(ctx, "serving on %s ...", l.Addr())
// start serving incoming connections
wg := sync.WaitGroup{}
......@@ -140,13 +141,13 @@ func (stor *Storage) Run(ctx context.Context) error {
// it tries to persist master link reconnecting as needed
//
// it always returns an error - either due to cancel or command from master to shutdown
func (stor *Storage) talkMaster(ctx context.Context) error {
// XXX errctx
func (stor *Storage) talkMaster(ctx context.Context) (err error) {
defer runningf(&ctx, "talk master(%v)", stor.node.MasterAddr)(&err)
for {
stor.vlogf("master(%v): connecting ...", stor.node.MasterAddr)
log.Info(ctx, "connecting ...")
err := stor.talkMaster1(ctx)
stor.logf("master(%v): %v", stor.node.MasterAddr, err)
log.Error(ctx, err)
// TODO if err = shutdown -> return
......@@ -187,7 +188,7 @@ func (stor *Storage) talkMaster1(ctx context.Context) (err error) {
// XXX -> node.Dial ?
if accept.YourNodeUUID != stor.node.MyInfo.NodeUUID {
stor.logf("%v: master told us to have UUID=%v", Mlink, accept.YourNodeUUID)
log.Infof(ctx, "%v: master told us to have UUID=%v", Mlink, accept.YourNodeUUID)
stor.node.MyInfo.NodeUUID = accept.YourNodeUUID
}
......@@ -225,13 +226,13 @@ func (stor *Storage) talkMaster1(ctx context.Context) (err error) {
// let master initialize us. If successful this ends with StartOperation command.
err = stor.m1initialize(ctx, Mconn)
if err != nil {
stor.logf("%v: master: %v", Mconn, err)
log.Error(ctx, err)
continue // retry initializing
}
// we got StartOperation command. Let master drive us during servicing phase.
err = stor.m1serve(ctx, Mconn)
stor.logf("%v: master: %v", Mconn, err)
log.Error(ctx, err)
continue // retry from initializing
}
......@@ -250,7 +251,7 @@ func (stor *Storage) talkMaster1(ctx context.Context) (err error) {
// - nil: initialization was ok and a command came from master to start operation
// - !nil: initialization was cancelled or failed somehow
func (stor *Storage) m1initialize(ctx context.Context, Mconn *neo.Conn) (err error) {
defer xerr.Context(&err, "init")
defer runningf(&ctx, "init %v", Mconn)(&err)
for {
msg, err := Mconn.Recv()
......@@ -319,7 +320,7 @@ func (stor *Storage) m1initialize(ctx context.Context, Mconn *neo.Conn) (err err
// either due to master commanding us to stop, or context cancel or some other
// error.
func (stor *Storage) m1serve(ctx context.Context, Mconn *neo.Conn) (err error) {
defer xerr.Context(&err, "serve")
defer runningf(&ctx, "serve %v", Mconn)(&err)
// refresh stor.opCtx and cancel it when we finish so that client
// handlers know they need to stop operating as master told us to do so.
......@@ -359,7 +360,7 @@ func (stor *Storage) m1serve(ctx context.Context, Mconn *neo.Conn) (err error) {
// ServeLink serves incoming node-node link connection
// XXX +error return?
func (stor *Storage) ServeLink(ctx context.Context, link *neo.NodeLink) {
stor.logf("%s: serving new node", link)
log.Infof(ctx, "%s: serving new node", link) // XXX -> running?
// close link when either cancelling or returning (e.g. due to an error)
// ( when cancelling - link.Close will signal to all current IO to
......@@ -374,7 +375,7 @@ func (stor *Storage) ServeLink(ctx context.Context, link *neo.NodeLink) {
// XXX ret err = ctx.Err()
case <-retch:
}
stor.logf("%v: closing link", link)
log.Info(ctx, "%v: closing link", link)
link.Close() // XXX err
}()
......@@ -382,7 +383,7 @@ func (stor *Storage) ServeLink(ctx context.Context, link *neo.NodeLink) {
// XXX only accept when operational (?)
nodeInfo, err := IdentifyPeer(link, neo.STORAGE)
if err != nil {
stor.logf("%v", err)
log.Error(ctx, err)
return
}
......@@ -393,7 +394,7 @@ func (stor *Storage) ServeLink(ctx context.Context, link *neo.NodeLink) {
default:
// XXX vvv should be reply to peer
stor.logf("%v: unexpected peer type: %v", link, nodeInfo.NodeType)
log.Errorf(ctx, "%v: unexpected peer type: %v", link, nodeInfo.NodeType)
return
}
......@@ -401,7 +402,7 @@ func (stor *Storage) ServeLink(ctx context.Context, link *neo.NodeLink) {
for {
conn, err := link.Accept()
if err != nil {
stor.logf("%v", err) // XXX err ctx
log.Error(ctx, err)
break
}
......@@ -427,7 +428,7 @@ func (stor *Storage) withWhileOperational(ctx context.Context) (context.Context,
// the connection is closed when serveClient returns
// XXX +error return?
func (stor *Storage) serveClient(ctx context.Context, conn *neo.Conn) {
stor.logf("%s: serving new client conn", conn)
log.Infof(ctx, "%s: serving new client conn", conn) // XXX -> running?
// rederive ctx to be also cancelled if M tells us StopOperation
ctx, cancel := stor.withWhileOperational(ctx)
......@@ -458,9 +459,9 @@ func (stor *Storage) serveClient(ctx context.Context, conn *neo.Conn) {
case err = <-done:
}
stor.logf("%v: %v", conn, err)
log.Infof(ctx, "%v: %v", conn, err)
// XXX vvv -> defer ?
stor.logf("%v: closing client conn", conn)
log.Infof(ctx, "%v: closing client conn", conn)
conn.Close() // XXX err
}
......
......@@ -19,3 +19,39 @@
package server
// misc utilities
import (
"context"
"fmt"
"lab.nexedi.com/kirr/neo/go/xcommon/task"
"lab.nexedi.com/kirr/neo/go/xcommon/log"
)
// running is syntatic sugar to push new task to operational stack, log it and
// adjust error return with task prefix.
//
// use like this:
//
// defer running(&ctx, "my task")(&err)
func running(ctxp *context.Context, name string) func(*error) {
return _running(ctxp, name)
}
// runningf is running cousing with formatting support
func runningf(ctxp *context.Context, format string, argv ...interface{}) func(*error) {
return _running(ctxp, fmt.Sprintf(format, argv...))
}
func _running(ctxp *context.Context, name string) func(*error) {
ctx := task.Running(*ctxp, name)
*ctxp = ctx
log.Depth(2).Info(ctx) // XXX level = ok?
return func(errp *error) {
// NOTE not *ctxp here - as context pointed by ctxp could be
// changed when this deferred function is run
task.ErrContext(errp, ctx)
// XXX also log task stop here?
}
}
......@@ -23,30 +23,79 @@
package log
import (
"context"
"fmt"
"github.com/golang/glog"
"lab.nexedi.com/kirr/neo/go/xcommon/task"
)
/*
// taskPrefix returns prefix associated to current operational task stack to put to logs
func taskPrefix(ctx) string {
func taskPrefix(ctx context.Context) string {
s := task.Current(ctx).String()
if s != "" {
s += ": "
}
return s
}
*/
// withTask prepends string describing current operational task stack to argv and returns it
// handy to use this way:
//
// func info(ctx, argv ...interface{}) {
// glog.Info(withTask(ctx, argv...)...)
// }
//
// see https://golang.org/issues/21388
func withTask(ctx context.Context, argv ...interface{}) []interface{} {
task := task.Current(ctx).String()
if task == "" {
return argv
}
if len(argv) != 0 {
task += ":"
}
return append([]interface{}{task}, argv...)
}
type Depth int
func (d Depth) Info(ctx context.Context, argv ...interface{}) {
// XXX avoid task formatting if logging severity disabled
glog.InfoDepth(int(d+1), withTask(ctx, argv...)...)
}
func (d Depth) Infof(ctx context.Context, format string, argv ...interface{}) {
// XXX avoid formatting if logging severity disables info
glog.InfoDepth(d+1, taskPrefix(ctx) + fmt.Sprintf(format, argv)
// XXX avoid formatting if logging severity disabled
glog.InfoDepth(int(d+1), withTask(ctx, fmt.Sprintf(format, argv))...)
}
func (d Depth) Error(ctx context.Context, argv ...interface{}) {
glog.ErrorDepth(int(d+1), withTask(ctx, argv...)...)
}
func (d Depth) Errorf(ctx context.Context, format string, argv ...interface{}) {
glog.ErrorDepth(int(d+1), withTask(ctx, fmt.Sprintf(format, argv))...)
}
func Info(ctx context.Context, argv ...interface{}) { Depth(1).Info(ctx, argv...) }
func Error(ctx context.Context, argv ...interface{}) { Depth(1).Error(ctx, argv...) }
func Infof(ctx context.Context, format string, argv ...interface{}) {
Depth(1).Infof(ctx, format, argv)
Depth(1).Infof(ctx, format, argv...)
}
func Errorf(ctx context.Context, format string, argv ...interface{}) {
Depth(1).Errorf(ctx, format, argv...)
}
......
......@@ -22,6 +22,9 @@ package task
import (
"context"
"fmt"
"lab.nexedi.com/kirr/go123/xerr"
)
// Task represents currently running operation
......@@ -34,13 +37,18 @@ type taskKey struct{}
// Running creates new task and returns new context with that task set to current
func Running(ctx context.Context, name string) context.Context {
context.WithValue(&Task{Parent: Current(ctx), Name: name})
return context.WithValue(ctx, taskKey{}, &Task{Parent: Current(ctx), Name: name})
}
// Runningf is Running cousing with formatting support
func Runningf(ctx context.Context, format string, argv ...interface{}) context.Context {
return Running(ctx, fmt.Sprintf(format, argv...))
}
// Current returns current task represented by context.
// if there is no current task - it returns nil.
func Current(ctx Context) *Task {
task, _ := ctx.Value(taskKey).(*Task)
func Current(ctx context.Context) *Task {
task, _ := ctx.Value(taskKey{}).(*Task)
return task
}
......@@ -49,16 +57,16 @@ func Current(ctx Context) *Task {
//
// func myfunc(ctx, ...) (..., err error) {
// ctx = task.Running("doing something")
// defer task.ErrContext(&err, ctx)
// ...
// defer task.ErrContext(&err, ctx)
// ...
//
// Please see lab.nexedi.com/kirr/go123/xerr.Context for semantic details
func ErrContext(errp *error, ctx Context) {
func ErrContext(errp *error, ctx context.Context) {
task := Current(ctx)
if task == nil {
return
}
return xerr.Context(errp, task.Name)
xerr.Context(errp, task.Name)
}
// String returns string representing whole operational stack.
......@@ -68,11 +76,11 @@ func ErrContext(errp *error, ctx Context) {
//
// nil Task is represented as ""
func (t *Task) String() string {
if o == nil {
if t == nil {
return ""
}
prefix := Parent.String()
prefix := t.Parent.String()
if prefix != "" {
prefix += ": "
}
......
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