lib9p

commit b55aa17fe41867d65e855ba8533527004308a0bd
parent 37d55bf6f90220be33d34b56d6ed12c25dcd11f9
Author: Matsuda Kenji <info@mtkn.jp>
Date:   Fri, 22 Dec 2023 09:25:04 +0900

attach pipeline

Diffstat:
M
server.go
 | 
140
+++++++++++++++++++++++++++++++++++++++++++++++++++----------------------------

1 file changed, 90 insertions(+), 50 deletions(-)
diff --git a/server.go b/server.go
@@ -234,6 +234,9 @@ func rVersion(ctx context.Context, c <-chan *Req) {
 		case <-ctx.Done():
 			return
 		case r := <-c:
+			if r == nil {
+				return
+			}
 			if r.err != nil {
 				panic(fmt.Errorf("rVersion err: %w", r.err))
 			}
@@ -283,6 +286,9 @@ func rAuth(ctx context.Context, c <-chan *Req) {
 		case <-ctx.Done():
 			return
 		case r := <-c:
+			if r == nil {
+				return
+			}
 			if r.err != nil {
 				r.Srv.fPool.delete(r.Ifcall.(*TAuth).Afid)
 				setError(r, r.err)
@@ -292,56 +298,87 @@ func rAuth(ctx context.Context, c <-chan *Req) {
 	}
 }
 
-func sAttach(ctx context.Context, s *Server, r *Req) {
-	ifcall := r.Ifcall.(*TAttach)
-	fid, err := s.fPool.add(ifcall.Fid)
-	if err != nil {
-		Respond(ctx, r, ErrDupFid)
-		return
-	}
-	switch {
-	case s.Auth == nil && ifcall.Afid == NOFID:
-	case s.Auth == nil && ifcall.Afid != NOFID:
-		Respond(ctx, r, ErrBotch)
-		return
-	case s.Auth != nil && ifcall.Afid == NOFID:
-		Respond(ctx, r, fmt.Errorf("authentication required"))
-		return
-	case s.Auth != nil && ifcall.Afid != NOFID:
-		afid, ok := s.fPool.lookup(ifcall.Afid)
-		if !ok {
-			Respond(ctx, r, ErrUnknownFid)
-			return
-		}
-		af, ok := afid.File.(*AuthFile)
-		if !ok {
-			log.Printf("afile: %[1]T, %[1]v", afid.File)
-			Respond(ctx, r, fmt.Errorf("not auth file"))
-			return
-		}
-		if !af.AuthOK {
-			Respond(ctx, r, fmt.Errorf("not authenticated"))
+func sAttach(ctx context.Context, s *Server, c <-chan *Req) {
+	rAttachChan := make(chan *Req)
+	defer close(rAttachChan)
+	go rAttach(ctx, rAttachChan)
+	for {
+		select {
+		case <-ctx.Done():
 			return
+		case r := <-c:
+			if r == nil {
+				return
+			}
+			ifcall := r.Ifcall.(*TAttach)
+			fid, err := s.fPool.add(ifcall.Fid)
+			if err != nil {
+				r.err = ErrDupFid
+				rAttachChan <- r
+				continue
+			}
+			switch {
+			case s.Auth == nil && ifcall.Afid == NOFID:
+			case s.Auth == nil && ifcall.Afid != NOFID:
+				r.err = ErrBotch
+				rAttachChan <- r
+				continue
+			case s.Auth != nil && ifcall.Afid == NOFID:
+				r.err = fmt.Errorf("authentication required")
+				rAttachChan <- r
+				continue
+			case s.Auth != nil && ifcall.Afid != NOFID:
+				afid, ok := s.fPool.lookup(ifcall.Afid)
+				if !ok {
+					r.err = ErrUnknownFid
+					rAttachChan <- r
+					continue
+				}
+				af, ok := afid.File.(*AuthFile)
+				if !ok {
+					log.Printf("afile: %[1]T, %[1]v", afid.File)
+					r.err = fmt.Errorf("not auth file")
+					rAttachChan <- r
+					continue
+				}
+				if !af.AuthOK {
+					r.err = fmt.Errorf("not authenticated")
+					rAttachChan <- r
+					continue
+				}
+			}
+			fid.OMode = -1
+			fid.path = "."
+			fid.Uid = ifcall.Uname
+			st, err := fs.Stat(ExportFS{s.fs}, ".")
+			if err != nil {
+				r.err = fmt.Errorf("stat root: %v", err)
+				rAttachChan <- r
+				continue
+			}
+			r.Ofcall = &RAttach{
+				Qid: st.Sys().(*Stat).Qid,
+			}
+			rAttachChan <- r
 		}
 	}
-	fid.OMode = -1
-	fid.path = "."
-	fid.Uid = ifcall.Uname
-	st, err := fs.Stat(ExportFS{s.fs}, ".")
-	if err != nil {
-		Respond(ctx, r, fmt.Errorf("stat root: %v", err))
-		return
-	}
-	r.Ofcall = &RAttach{
-		Qid: st.Sys().(*Stat).Qid,
-	}
-	Respond(ctx, r, nil)
 }
 
-func rAttach(r *Req, err error) {
-	if err != nil {
-		r.Srv.fPool.delete(r.Ifcall.(*TAttach).Fid)
-		setError(r, err)
+func rAttach(ctx context.Context, c <-chan *Req) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case r := <- c:
+			if r == nil {
+				return
+			}
+			if r.err != nil {
+				r.Srv.fPool.delete(r.Ifcall.(*TAttach).Fid)
+				setError(r, r.err)
+			}
+			r.Srv.respChan <- r
+		}
 	}
 }
 
@@ -981,14 +1018,19 @@ func (s *Server) Serve(ctx context.Context) {
 	defer cancel()
 	s.runListener(ctx)
 	s.runSpeaker(ctx)
-	versionChan := make(chan *Req)
-	authChan := make(chan *Req)
+	var (
+		versionChan = make(chan *Req)
+		authChan = make(chan *Req)
+		attachChan = make(chan *Req)
+	)
 	defer func() {
 		close(versionChan)
 		close(authChan)
+		close(attachChan)
 	}()
 	go sVersion(ctx, s, versionChan)
 	go sAuth(ctx, s, authChan)
+	go sAttach(ctx, s, attachChan)
 	go Respond2(ctx, s)
 L:
 	for {
@@ -1012,7 +1054,7 @@ L:
 				case *TAuth:
 					authChan <- r
 				case *TAttach:
-					sAttach(ctx1, s, r)
+					attachChan <- r
 				case *TFlush:
 					sFlush(ctx1, s, r)
 				case *TWalk:
@@ -1049,8 +1091,6 @@ func Respond(ctx context.Context, r *Req, err error) {
 	switch r.Ifcall.(type) {
 	default:
 		panic(fmt.Errorf("bug: r.Ifcall: %v", r.Ifcall))
-	case *TAttach:
-		rAttach(r, err)
 	case *TFlush:
 		rFlush(r, err)
 	case *TWalk: