feat: 为 Python worker 输出添加行前缀,并启用无缓冲模式
- 新增 prefixWriter,对每行输出添加 [python:N] 前缀,便于与 Go 日志区分 - 注入 PYTHONUNBUFFERED=1,确保 print() 实时输出不被管道缓冲
This commit is contained in:
52
prefix_writer.go
Normal file
52
prefix_writer.go
Normal file
@@ -0,0 +1,52 @@
|
|||||||
|
package gobridge
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"io"
|
||||||
|
"sync"
|
||||||
|
)
|
||||||
|
|
||||||
|
// prefixWriter 对每一行输出添加固定前缀,线程安全。
|
||||||
|
type prefixWriter struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
w io.Writer
|
||||||
|
prefix []byte
|
||||||
|
buf []byte
|
||||||
|
}
|
||||||
|
|
||||||
|
func newPrefixWriter(w io.Writer, prefix string) *prefixWriter {
|
||||||
|
return &prefixWriter{w: w, prefix: []byte(prefix)}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *prefixWriter) Write(b []byte) (int, error) {
|
||||||
|
p.mu.Lock()
|
||||||
|
defer p.mu.Unlock()
|
||||||
|
|
||||||
|
n := len(b)
|
||||||
|
p.buf = append(p.buf, b...)
|
||||||
|
|
||||||
|
for {
|
||||||
|
idx := bytes.IndexByte(p.buf, '\n')
|
||||||
|
if idx < 0 {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
line := append(p.prefix, p.buf[:idx+1]...)
|
||||||
|
if _, err := p.w.Write(line); err != nil {
|
||||||
|
return n, err
|
||||||
|
}
|
||||||
|
p.buf = p.buf[idx+1:]
|
||||||
|
}
|
||||||
|
return n, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// flush 将缓冲区中尚未以换行结尾的内容输出(进程退出时调用)。
|
||||||
|
func (p *prefixWriter) flush() {
|
||||||
|
p.mu.Lock()
|
||||||
|
defer p.mu.Unlock()
|
||||||
|
if len(p.buf) > 0 {
|
||||||
|
line := append(p.prefix, p.buf...)
|
||||||
|
line = append(line, '\n')
|
||||||
|
p.w.Write(line) //nolint
|
||||||
|
p.buf = nil
|
||||||
|
}
|
||||||
|
}
|
||||||
18
worker.go
18
worker.go
@@ -60,18 +60,20 @@ func (w *worker) start() error {
|
|||||||
fmt.Sprintf("GOBRIDGE_WORKER_ID=%d", w.id),
|
fmt.Sprintf("GOBRIDGE_WORKER_ID=%d", w.id),
|
||||||
fmt.Sprintf("GOBRIDGE_WORKER_COUNT=%d", w.cfg.workers),
|
fmt.Sprintf("GOBRIDGE_WORKER_COUNT=%d", w.cfg.workers),
|
||||||
"GOBRIDGE_DEATH_FD=3", // ExtraFiles[0] → fd 3
|
"GOBRIDGE_DEATH_FD=3", // ExtraFiles[0] → fd 3
|
||||||
|
"PYTHONUNBUFFERED=1", // Python stdout/stderr 不缓冲,print() 立即输出
|
||||||
)
|
)
|
||||||
cmd.ExtraFiles = []*os.File{pr} // fd 3 in child
|
cmd.ExtraFiles = []*os.File{pr} // fd 3 in child
|
||||||
if w.cfg.stdout != nil {
|
prefix := fmt.Sprintf("[python:%d] ", w.id)
|
||||||
cmd.Stdout = w.cfg.stdout
|
stdout := w.cfg.stdout
|
||||||
} else {
|
if stdout == nil {
|
||||||
cmd.Stdout = os.Stdout
|
stdout = os.Stdout
|
||||||
}
|
}
|
||||||
if w.cfg.stderr != nil {
|
stderr := w.cfg.stderr
|
||||||
cmd.Stderr = w.cfg.stderr
|
if stderr == nil {
|
||||||
} else {
|
stderr = os.Stderr
|
||||||
cmd.Stderr = os.Stderr
|
|
||||||
}
|
}
|
||||||
|
cmd.Stdout = newPrefixWriter(stdout, prefix)
|
||||||
|
cmd.Stderr = newPrefixWriter(stderr, prefix)
|
||||||
|
|
||||||
if err := cmd.Start(); err != nil {
|
if err := cmd.Start(); err != nil {
|
||||||
pr.Close()
|
pr.Close()
|
||||||
|
|||||||
Reference in New Issue
Block a user