frankenphp/worker.go

package frankenphp

// #include <stdlib.h>
// #include "frankenphp.h"
import "C"
import (
	"errors"
	"fmt"
	"net/http"
	"path/filepath"
	"runtime/cgo"
	"sync"

	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
)

var (
	workersRequestChans sync.Map // map[fileName]chan *http.Request
	workersReadyWG      sync.WaitGroup
)

// TODO: start all the worker in parallell to reduce the boot time
func initWorkers(opt []workerOpt) error {
	for _, w := range opt {
		if err := startWorkers(w.fileName, w.num, w.env); err != nil {
			return err
		}
	}

	return nil
}

func startWorkers(fileName string, nbWorkers int, env PreparedEnv) error {
	absFileName, err := filepath.Abs(fileName)
	if err != nil {
		return fmt.Errorf("workers %q: %w", fileName, err)
	}

	if _, ok := workersRequestChans.Load(absFileName); ok {
		return fmt.Errorf("workers %q: already started", absFileName)
	}

	workersRequestChans.Store(absFileName, make(chan *http.Request))
	shutdownWG.Add(nbWorkers)
	workersReadyWG.Add(nbWorkers)

	var (
		m    sync.RWMutex
		errs []error
	)

	if env == nil {
		env = make(PreparedEnv, 1)
	}

	env["FRANKENPHP_WORKER\x00"] = "1"

	l := getLogger()
	for i := 0; i < nbWorkers; i++ {
		go func() {
			defer shutdownWG.Done()
			for {
				// Create main dummy request
				r, err := http.NewRequest(http.MethodGet, filepath.Base(absFileName), nil)

				metrics.StartWorker(absFileName)

				if err != nil {
					panic(err)
				}
				r, err = NewRequestWithContext(
					r,
					WithRequestDocumentRoot(filepath.Dir(absFileName), false),
					WithRequestPreparedEnv(env),
				)
				if err != nil {
					panic(err)
				}

				if c := l.Check(zapcore.DebugLevel, "starting"); c != nil {
					c.Write(zap.String("worker", absFileName), zap.Int("num", nbWorkers))
				}

				if err := ServeHTTP(nil, r); err != nil {
					panic(err)
				}

				fc := r.Context().Value(contextKey).(*FrankenPHPContext)
				if fc.currentWorkerRequest != 0 {
					// Terminate the pending HTTP request handled by the worker
					maybeCloseContext(fc.currentWorkerRequest.Value().(*http.Request).Context().Value(contextKey).(*FrankenPHPContext))
					fc.currentWorkerRequest.Delete()
					fc.currentWorkerRequest = 0
				}

				// TODO: make the max restart configurable
				if _, ok := workersRequestChans.Load(absFileName); ok {
					metrics.StopWorker(absFileName)
					workersReadyWG.Add(1)
					if fc.exitStatus == 0 {
						if c := l.Check(zapcore.InfoLevel, "restarting"); c != nil {
							c.Write(zap.String("worker", absFileName))
						}
					} else {
						if c := l.Check(zapcore.ErrorLevel, "unexpected termination, restarting"); c != nil {
							c.Write(zap.String("worker", absFileName), zap.Int("exit_status", int(fc.exitStatus)))
						}
					}
				} else {
					break
				}
			}

			// TODO: check if the termination is expected
			if c := l.Check(zapcore.DebugLevel, "terminated"); c != nil {
				c.Write(zap.String("worker", absFileName))
			}
		}()
	}

	workersReadyWG.Wait()
	m.Lock()
	defer m.Unlock()

	if len(errs) == 0 {
		return nil
	}

	return fmt.Errorf("workers %q: error while starting: %w", fileName, errors.Join(errs...))
}

func stopWorkers() {
	workersRequestChans.Range(func(k, v any) bool {
		workersRequestChans.Delete(k)

		return true
	})
}

//export go_frankenphp_worker_ready
func go_frankenphp_worker_ready() {
	workersReadyWG.Done()
}

//export go_frankenphp_worker_handle_request_start
func go_frankenphp_worker_handle_request_start(mrh C.uintptr_t) C.uintptr_t {
	mainRequest := cgo.Handle(mrh).Value().(*http.Request)
	fc := mainRequest.Context().Value(contextKey).(*FrankenPHPContext)

	v, ok := workersRequestChans.Load(fc.scriptFilename)
	if !ok {
		// Probably shutting down
		return 0
	}

	rc := v.(chan *http.Request)
	l := getLogger()

	if c := l.Check(zapcore.DebugLevel, "waiting for request"); c != nil {
		c.Write(zap.String("worker", fc.scriptFilename))
	}

	var r *http.Request
	select {
	case <-done:
		if c := l.Check(zapcore.DebugLevel, "shutting down"); c != nil {
			c.Write(zap.String("worker", fc.scriptFilename))
		}

		return 0
	case r = <-rc:
	}

	fc.currentWorkerRequest = cgo.NewHandle(r)
	r.Context().Value(handleKey).(*handleList).AddHandle(fc.currentWorkerRequest)

	if c := l.Check(zapcore.DebugLevel, "request handling started"); c != nil {
		c.Write(zap.String("worker", fc.scriptFilename), zap.String("url", r.RequestURI))
	}

	if err := updateServerContext(r, false, mrh); err != nil {
		// Unexpected error
		if c := l.Check(zapcore.DebugLevel, "unexpected error"); c != nil {
			c.Write(zap.String("worker", fc.scriptFilename), zap.String("url", r.RequestURI), zap.Error(err))
		}

		return 0
	}

	return C.uintptr_t(fc.currentWorkerRequest)
}

//export go_frankenphp_finish_request
func go_frankenphp_finish_request(mrh, rh C.uintptr_t, deleteHandle bool) {
	rHandle := cgo.Handle(rh)
	r := rHandle.Value().(*http.Request)
	fc := r.Context().Value(contextKey).(*FrankenPHPContext)

	if deleteHandle {
		r.Context().Value(handleKey).(*handleList).FreeAll()
		cgo.Handle(mrh).Value().(*http.Request).Context().Value(contextKey).(*FrankenPHPContext).currentWorkerRequest = 0
	}

	maybeCloseContext(fc)

	if c := fc.logger.Check(zapcore.DebugLevel, "request handling finished"); c != nil {
		var fields []zap.Field
		if mrh == 0 {
			fields = append(fields, zap.String("worker", fc.scriptFilename), zap.String("url", r.RequestURI))
		} else {
			fields = append(fields, zap.String("url", r.RequestURI))
		}

		c.Write(fields...)
	}
}