lots of fixes, but it's still broken because of filemarshal issues

fritz committed 14 years ago

Commit d85aa851 parent 7b395d07 files · +61 −71Browse files

Unified · Split

7 files

common.go +22 −29
main.go +17 −17
master.go +3 −6
mexec.go +10 −9
misc.go +4 −5
slave.go +2 −2
web.go +3 −3

@@ -10,13 +10,13 @@ · 9 unmodified lines
10	package main	10	package main
11		11
12	import (	12	import (
13	"os"	13	"bitbucket.org/floren/filemarshal"
14	"net"	14	"encoding/gob"
15	"fmt"	15	"fmt"
16	"log"
17	"io"	16	"io"
18	"gob"	17	"log"
19	"bitbucket.org/floren/filemarshal"	18	"net"
		19	"os"
20	"strconv"	20	"strconv"
21	"strings"	21	"strings"
22	"syscall"	22	"syscall"
@@ -30,7 +30,6 @@ func (s SlaveResp) String() string { · 7 unmodified lines
30	return fmt.Sprint("id: ", s.Id)	30	return fmt.Sprint("id: ", s.Id)
31	}	31	}
32		32
33
34	type Resp struct {	33	type Resp struct {
35	NumNodes int	34	NumNodes int
36	Msg string	35	Msg string
@@ -52,7 +51,7 @@ type cmdToExec struct { · 15 unmodified lines
52	DestName string	51	DestName string
53	SymlinkTarget string	52	SymlinkTarget string
54	Local int	53	Local int
55	Fi *os.FileInfo	54	Fi os.FileInfo
56	}	55	}
57		56
58	func (a *cmdToExec) String() string {	57	func (a *cmdToExec) String() string {
@@ -72,7 +71,7 @@ type vitalData struct { · 13 unmodified lines
72	HostAddr string	71	HostAddr string
73	ParentAddr string	72	ParentAddr string
74	ServerAddr string	73	ServerAddr string
75	Id string	74	Id string
76	Nodes []string	75	Nodes []string
77	Exceptlist map[string]bool	76	Exceptlist map[string]bool
78	}	77	}
@@ -144,7 +143,6 @@ func (s *SlaveInfo) String() string { · 65 unmodified lines
144	return fmt.Sprint(s.Id, "@", s.Addr)	143	return fmt.Sprint(s.Id, "@", s.Addr)
145	}	144	}
146		145
147
148	func Dprint(level int, arg ...interface{}) {	146	func Dprint(level int, arg ...interface{}) {
149	if *DebugLevel >= level {	147	if *DebugLevel >= level {
150	log.Print(arg...)	148	log.Print(arg...)
@@ -168,7 +166,6 @@ const ( · 17 unmodified lines
168	Recv	166	Recv
169	)	167	)
170		168
171
172	func IoString(i interface{}, dir int) string {	169	func IoString(i interface{}, dir int) string {
173	switch i.(type) {	170	switch i.(type) {
174	case net.Conn:	171	case net.Conn:
@@ -227,7 +224,7 @@ func (r *RpcClientServer) Send(funcname string, arg interface{}) { · 52 unmodified lines
227		224
228	var onRecvFunc func(funcname string, r io.Reader, arg interface{})	225	var onRecvFunc func(funcname string, r io.Reader, arg interface{})
229		226
230	func (r *RpcClientServer) Recv(funcname string, arg interface{}) (err os.Error) {	227	func (r *RpcClientServer) Recv(funcname string, arg interface{}) (err error) {
231	err = r.D.Decode(arg)	228	err = r.D.Decode(arg)
232	if err != nil {	229	if err != nil {
233	log.Print(funcname, ": Recv error: ", err)	230	log.Print(funcname, ": Recv error: ", err)
@@ -242,10 +239,9 @@ func (r *RpcClientServer) Recv(funcname string, arg interface{}) (err os.Error) · 8 unmodified lines
242	return	239	return
243	}	240	}
244		241
245
246	var onDialFunc func(fam, laddr, raddr string)	242	var onDialFunc func(fam, laddr, raddr string)
247		243
248	func Dial(fam, laddr, raddr string) (c net.Conn, err os.Error) {	244	func Dial(fam, laddr, raddr string) (c net.Conn, err error) {
249	if onDialFunc != nil {	245	if onDialFunc != nil {
250	onDialFunc(fam, laddr, raddr)	246	onDialFunc(fam, laddr, raddr)
251	}	247	}
@@ -265,7 +261,6 @@ func Dial(fam, laddr, raddr string) (c net.Conn, err os.Error) { · 13 unmodified lines
265	return	261	return
266	}	262	}
267		263
268
269	type Listener struct {	264	type Listener struct {
270	l net.Listener	265	l net.Listener
271	}	266	}
@@ -276,7 +271,7 @@ func (l Listener) Addr() net.Addr { · 4 unmodified lines
276		271
277	var onListenFunc func(fam, laddr string)	272	var onListenFunc func(fam, laddr string)
278		273
279	func Listen(fam, laddr string) (l Listener, err os.Error) {	274	func Listen(fam, laddr string) (l Listener, err error) {
280	if onListenFunc != nil {	275	if onListenFunc != nil {
281	onListenFunc(fam, laddr)	276	onListenFunc(fam, laddr)
282	}	277	}
@@ -287,7 +282,7 @@ func Listen(fam, laddr string) (l Listener, err os.Error) { · 4 unmodified lines
287		282
288	var onAcceptFunc func(c net.Conn)	283	var onAcceptFunc func(c net.Conn)
289		284
290	func (l Listener) Accept() (c net.Conn, err os.Error) {	285	func (l Listener) Accept() (c net.Conn, err error) {
291	c, err = l.l.Accept()	286	c, err = l.l.Accept()
292	if err != nil {	287	if err != nil {
293	return	288	return
@@ -299,13 +294,12 @@ func (l Listener) Accept() (c net.Conn, err os.Error) { · 5 unmodified lines
299	return	294	return
300	}	295	}
301		296
302
303	// depends on syscall	297	// depends on syscall
304	func WaitAllChildren() {	298	func WaitAllChildren() {
305	var status syscall.WaitStatus	299	var status syscall.WaitStatus
306	for {	300	for {
307	pid, err := syscall.Wait4(-1, &status, 0, nil)	301	pid, err := syscall.Wait4(-1, &status, 0, nil)
308	if err <= 0 {	302	if err != nil {
309	break	303	break
310	}	304	}
311	log.Printf("wait4 returns pid %v status %v\n", pid, status)	305	log.Printf("wait4 returns pid %v status %v\n", pid, status)
@@ -318,7 +312,7 @@ func WaitAllChildren() { · 6 unmodified lines
318	* This function builds up a list of files that need to go out to the current node's sub-nodes.	312	* This function builds up a list of files that need to go out to the current node's sub-nodes.
319	* It is called by both the master and, if a more complex hierarchy is used, the upper-level slaves.	313	* It is called by both the master and, if a more complex hierarchy is used, the upper-level slaves.
320	*/	314	*/
321	func cacheRelayFilesAndDelegateExec(arg *StartReq, root, clientnode string) os.Error {	315	func cacheRelayFilesAndDelegateExec(arg *StartReq, root, clientnode string) error {
322	Dprint(2, "cacheRelayFilesAndDelegateExec: files ", arg.Cmds, " nodes: ", clientnode, " fileServer: ", arg.Lfam, arg.Lserver)	316	Dprint(2, "cacheRelayFilesAndDelegateExec: files ", arg.Cmds, " nodes: ", clientnode, " fileServer: ", arg.Lfam, arg.Lserver)
323		317
324	larg := newStartReq(arg)	318	larg := newStartReq(arg)
@@ -328,7 +322,7 @@ func cacheRelayFilesAndDelegateExec(arg *StartReq, root, clientnode string) os.E · 3 unmodified lines
328	comesfrom := root + c.DestName	322	comesfrom := root + c.DestName
329	Dprint(2, "current cmd comesfrom = ", comesfrom, ", DestName = ", c.DestName, ", CurrentName = ", c.CurrentName, ", SymlinkTarget = ", c.SymlinkTarget)	323	Dprint(2, "current cmd comesfrom = ", comesfrom, ", DestName = ", c.DestName, ", CurrentName = ", c.CurrentName, ", SymlinkTarget = ", c.SymlinkTarget)
330	f := new(filemarshal.File)	324	f := new(filemarshal.File)
331	if c.Fi.IsRegular() \|\| c.Fi.IsDirectory() \|\| c.Fi.IsSymlink() {	325	if !c.Fi.IsDir() \|\| c.Fi.IsDir() \|\| ((c.Fi.Mode() & os.ModeSymlink) != 0) {
332	f = &filemarshal.File{CurrentName: comesfrom, Fi: *c.Fi, SymlinkTarget: c.SymlinkTarget, DestName: c.DestName}	326	f = &filemarshal.File{CurrentName: comesfrom, Fi: *c.Fi, SymlinkTarget: c.SymlinkTarget, DestName: c.DestName}
333	} else {	327	} else {
334	continue	328	continue
@@ -360,6 +354,7 @@ func cacheRelayFilesAndDelegateExec(arg *StartReq, root, clientnode string) os.E · 25 unmodified lines
360		354
361	return nil	355	return nil
362	}	356	}
		357
363	/*	358	/*
364	* The ioProxy listens for incoming connections. Sub-nodes will connect to it	359	* The ioProxy listens for incoming connections. Sub-nodes will connect to it
365	* and use the connection as stdin, stdout, and stderr for the programs they	360	* and use the connection as stdin, stdout, and stderr for the programs they
@@ -371,7 +366,7 @@ func cacheRelayFilesAndDelegateExec(arg *StartReq, root, clientnode string) os.E · 5 unmodified lines
371	* finished. workerChan will contain one int for every client which has	366	* finished. workerChan will contain one int for every client which has
372	* completed and disconnected.	367	* completed and disconnected.
373	*/	368	*/
374	func ioProxy(fam, server string, dest io.Writer) (workerChan chan int, l Listener, err os.Error) {	369	func ioProxy(fam, server string, dest io.Writer) (workerChan chan int, l Listener, err error) {
375	workerChan = make(chan int, 0)	370	workerChan = make(chan int, 0)
376	l, err = Listen(fam, server)	371	l, err = Listen(fam, server)
377	if err != nil {	372	if err != nil {
@@ -416,7 +411,7 @@ type nodeExecList struct { · 38 unmodified lines
416	* the really hard way: few people understand regular expressions. So we	411	* the really hard way: few people understand regular expressions. So we
417	* don't use them.	412	* don't use them.
418	*/	413	*/
419	func parseNodeList(l string) (rl []nodeExecList, err os.Error) {	414	func parseNodeList(l string) (rl []nodeExecList, err error) {
420	/* bust it apart by , */	415	/* bust it apart by , */
421	ranges := strings.SplitN(l, ",", -1)	416	ranges := strings.SplitN(l, ",", -1)
422	for _, n := range ranges {	417	for _, n := range ranges {
@@ -465,7 +460,7 @@ func doPrivateMount(pathbase string) { · 42 unmodified lines
465	* Make a TCP connection to a server, return both the connection and	460	* Make a TCP connection to a server, return both the connection and
466	* a File pointing to that connection.	461	* a File pointing to that connection.
467	*/	462	*/
468	func fileTcpDial(server string) (*os.File, net.Conn, os.Error) {	463	func fileTcpDial(server string) (*os.File, net.Conn, error) {
469	var laddr net.TCPAddr	464	var laddr net.TCPAddr
470	raddr, err := net.ResolveTCPAddr("tcp4", server)	465	raddr, err := net.ResolveTCPAddr("tcp4", server)
471	if err != nil {	466	if err != nil {
@@ -503,13 +498,11 @@ func newStartReq(arg StartReq) StartReq { · 31 unmodified lines
503	}	498	}
504	}	499	}
505		500
506
507	/*	501	/*
508	* Functions and data types for keeping track of slave nodes	502	* Functions and data types for keeping track of slave nodes
509	*/	503	*/
510		504
511		505	func registerSlaves() error {
512	func registerSlaves() os.Error {
513	l, err := Listen(*defaultFam, myListenAddress)	506	l, err := Listen(*defaultFam, myListenAddress)
514	if err != nil {	507	if err != nil {
515	log.Fatal("listen error:", err)	508	log.Fatal("listen error:", err)
@@ -584,8 +577,8 @@ func (sv Slaves) Add(vd vitalData, r *RpcClientServer) (resp SlaveResp) { · 68 unmodified lines
584		577
585	func (sv Slaves) Remove(s SlaveInfo) {	578	func (sv Slaves) Remove(s SlaveInfo) {
586	Dprintln(4, "Remove %v ", s, " slave %v", sv.Slaves[s.Id])	579	Dprintln(4, "Remove %v ", s, " slave %v", sv.Slaves[s.Id])
587	sv.Slaves[s.Id] = nil, false	580	delete(sv.Slaves, s.Id)
588	sv.Addr2id[s.Server] = "", false	581	delete(sv.Addr2id, s.Server)
589	Dprintln(2, "slave Remove: Id: ", s)	582	Dprintln(2, "slave Remove: Id: ", s)
590	return	583	return
591	}	584	}
@@ -600,7 +593,7 @@ func (sv Slaves) Remove(s SlaveInfo) { · 8 unmodified lines
600	func (sv Slaves) Get(n string) (s SlaveInfo, ok bool) {	593	func (sv Slaves) Get(n string) (s SlaveInfo, ok bool) {
601	Dprint(6, "Get: ", n)	594	Dprint(6, "Get: ", n)
602	s, ok = sv.Slaves[n]	595	s, ok = sv.Slaves[n]
603	if ! ok {	596	if !ok {
604	s, ok = sv.Slaves[sv.Addr2id[n]]	597	s, ok = sv.Slaves[sv.Addr2id[n]]
605	}	598	}
606	Dprint(6, " Returns: ", s)	599	Dprint(6, " Returns: ", s)

main.go+17−17

 package main
 import (
+	"bitbucket.org/rminnich/forth"
+	"encoding/gob"
+	"flag"
+	"fmt"
 	"log"
+	"net/rpc"
 	"os"
-	"rpc"
-	"fmt"
 	"strconv"
-	"flag"
-	"gob"
-	"bitbucket.org/rminnich/forth"
 )
 func usage() {
 	defaultMasterUDS = flag.String("defaultMasterUDS", "/tmp/g", "Default Master Unix Domain Socket")
 	ioProxyPort      = flag.String("iopp", "0", "io proxy port")
 	cmdPort          = flag.String("cmdport", "6666", "command port")
-	defaultFam = flag.String("fam", "tcp4", "network type")
+	defaultFam       = flag.String("fam", "tcp4", "network type")
-	gprocBin	= flag.String("gprocBin", "gproc", "name of gproc binary")
+	gprocBin         = flag.String("gprocBin", "gproc", "name of gproc binary")
 	/* required in the command line */
-	parent           = flag.String("parent", "hostname", "parent for some configurations")
+	parent    = flag.String("parent", "hostname", "parent for some configurations")
 	myAddress = flag.String("myAddress", "hostname", "Required set to my address")
-	myId = flag.String("myId", "0", "Required -- tell slaves their id")
+	myId      = flag.String("myId", "0", "Required -- tell slaves their id")
 	/* these are not switches */
-	role = "client"
+	role            = "client"
 	myListenAddress string
 )
  * You can just put a simple string in for the argument and it will push and pop it
  *  ./gproc_linux_arm -parent='hostname base 7 roundup sb strcat 10.1.1.1 hostname base 7 % ifelse' -myId='hostname base 7 % 1  + hostname base 7 / hostname base 7 % ifelse' -myAddress=hostname s
  * is used for strongbox, where there is a single root, 49 level 1s defined by name %7 == 0, and
  */
 func main() {
-	var err os.Error
+	var err error
 	flag.Usage = usage
 	flag.Parse()
 	interp := forth.New()
 		info := getInfo(*defaultMasterUDS, flag.Arg(1))
 		fmt.Print("Nodes:\n", info)
 		/* not yet
-		 case "EXCEPT", "except", "x":
+		case "EXCEPT", "except", "x":
-		 loc.Init("init")
+		loc.Init("init")
-		 exceptOK := except(*defaultMasterUDS, flag.Args()[1:])
+		exceptOK := except(*defaultMasterUDS, flag.Args()[1:])
-		 fmt.Print(exceptOK)
+		fmt.Print(exceptOK)
-		 */
+		*/
 	case "R":
 		/* This is for executing a program from the slave */
 		slaveProc(NewRpcClientServer(os.Stdin, *binRoot), &RpcClientServer{E: gob.NewEncoder(os.Stdout), D: gob.NewDecoder(os.Stdout)}, &RpcClientServer{E: gob.NewEncoder(os.NewFile(3, "pipe")), D: gob.NewDecoder(os.NewFile(3, "pipe"))})

master.go+3−6

 package main
-import (
+import "log"
-	"os"
-	"log"
-)
 var (
 	Workers     []Worker
 	slaveNodes, err := parseNodeList(sendReq.Nodes)
 	Dprint(2, "receiveCmds: sendReq.Nodes: ", sendReq.Nodes, " expands to ", slaveNodes)
 	if err != nil {
-		r.Send("receiveCmds", Resp{NumNodes: 0, Msg: "startExecution: bad slaveNodeList: " + err.String()})
+		r.Send("receiveCmds", Resp{NumNodes: 0, Msg: "startExecution: bad slaveNodeList: " + err.Error()})
 		return
 	}
 	for _, aNode := range slaveNodes {
 /*
  * The master sits in a loop listening for commands to come in over the Unix domain socket.
  */
-func receiveCmds(domainSock string) os.Error {
+func receiveCmds(domainSock string) error {
 	vitalData := vitalData{HostAddr: "", HostReady: false, Error: "No hosts ready", Exceptlist: exceptFiles}
 	l, err := Listen("unix", *defaultMasterUDS)
 	if err != nil {

mexec.go+10−9

 package main
 import (
+	"bitbucket.org/rminnich/ldd"
+	"errors"
+	"fmt"
 	"log"
 	"os"
-	"strings"
-	"bitbucket.org/rminnich/ldd"
 	"path"
 	"path/filepath"
-	"fmt"
+	"strings"
 )
 /*
 }
 var (
-	BadRangeErr = os.NewError("bad range format")
+	BadRangeErr = errors.New("bad range format")
 )
 type packVisitor struct {
 	return &packVisitor{alreadyVisited: make(map[string]bool)}
 }
-func (p *packVisitor) VisitDir(filePath string, f *os.FileInfo) bool {
+func (p *packVisitor) VisitDir(filePath string, f os.FileInfo) bool {
 	filePath = strings.TrimSpace(filePath)
 	filePath = strings.TrimRightFunc(filePath, isNull)
 	return r == 0
 }
-func (p *packVisitor) VisitFile(filePath string, f *os.FileInfo) {
+func (p *packVisitor) VisitFile(filePath string, f os.FileInfo) {
 	// shouldn't need to do this, need to fix ldd
 	filePath = strings.TrimSpace(filePath)
 	filePath = strings.TrimRightFunc(filePath, isNull)
 		Local:       0,
 		Fi:          f,
 	}
-	Dprint(4, "VisitFile: appending ", f.Name, " ", f.Size, " ", []byte(filePath), " ", p.alreadyVisited)
+	Dprint(4, "VisitFile: appending ", f.Name(), " ", f.Size(), " ", []byte(filePath), " ", p.alreadyVisited)
 	p.cmds = append(p.cmds, c)
 	switch {
-	case f.IsRegular():
+	case !f.IsDir():
-		p.bytesToTransfer += f.Size
+		p.bytesToTransfer += f.Size()
 	case f.IsSymlink():
 		/* we have to read the link but also get a path the file for
 		 * further walking. We think.

misc.go+4−5

@@ -31,11 +31,10 @@ import ( · 30 unmodified lines
31	// return cmds	31	// return cmds
32	// }	32	// }
33		33
34		34	func readitin(s, root string) ([]byte, os.FileInfo, error) {
35	func readitin(s, root string) ([]byte, os.FileInfo, os.Error) {
36	fi, _ := os.Stat(root + s)	35	fi, _ := os.Stat(root + s)
37	f, _ := os.Open(s)	36	f, _ := os.Open(s)
38	bytes := make([]byte, fi.Size)	37	bytes := make([]byte, fi.Size())
39	f.Read(bytes)	38	f.Read(bytes)
40	return bytes, *fi, nil	39	return bytes, *fi, nil
41	}	40	}
@@ -44,12 +43,12 @@ type Arg struct { · 2 unmodified lines
44	Msg []byte	43	Msg []byte
45	}	44	}
46		45
47	func Ping(arg Arg, resp Resp) os.Error {	46	func Ping(arg Arg, resp Resp) error {
48	resp.Msg = string(arg.Msg)	47	resp.Msg = string(arg.Msg)
49	return nil	48	return nil
50	}	49	}
51		50
52	func Debug(arg SetDebugLevel, resp SetDebugLevel) os.Error {	51	func Debug(arg SetDebugLevel, resp SetDebugLevel) error {
53	resp.level = *DebugLevel	52	resp.level = *DebugLevel
54	*DebugLevel = arg.level	53	*DebugLevel = arg.level
55	return nil	54	return nil

slave.go+2−2

@@ -15,9 +15,9 @@ import ( · 14 unmodified lines
15	"os"	15	"os"
16	"net"	16	"net"
17	"fmt"	17	"fmt"
18	"gob"	18	"encoding/gob"
19	"time"	19	"time"
20	"rand"	20	"math/rand"
21	)	21	)
22		22
23	var id string	23	var id string

web.go+3−3

@@ -10,13 +10,13 @@ · 9 unmodified lines
10	package main	10	package main
11		11
12	import (	12	import (
13	"http"	13	"net/http"
14	"io"	14	"io"
15	"log"	15	"log"
16	"old/template"	16	"old/template"
17	"exec"	17	"os/exec"
18	"os"	18	"os"
19	"url"	19	"net/url"
20	)	20	)
21		21
22	var fmap = template.FormatterMap{	22	var fmap = template.FormatterMap{