worker.go 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187
  1. package frankenphp
  2. // #include <stdlib.h>
  3. // #include "frankenphp.h"
  4. import "C"
  5. import (
  6. "errors"
  7. "fmt"
  8. "net/http"
  9. "path/filepath"
  10. "runtime/cgo"
  11. "sync"
  12. "go.uber.org/zap"
  13. )
  14. var (
  15. workersRequestChans sync.Map // map[fileName]chan *http.Request
  16. workersReadyWG sync.WaitGroup
  17. )
  18. // TODO: start all the worker in parallell to reduce the boot time
  19. func initWorkers(opt []workerOpt) error {
  20. for _, w := range opt {
  21. if err := startWorkers(w.fileName, w.num, w.env); err != nil {
  22. return err
  23. }
  24. }
  25. return nil
  26. }
  27. func startWorkers(fileName string, nbWorkers int, env map[string]string) error {
  28. absFileName, err := filepath.Abs(fileName)
  29. if err != nil {
  30. return fmt.Errorf("workers %q: %w", fileName, err)
  31. }
  32. if _, ok := workersRequestChans.Load(absFileName); ok {
  33. return fmt.Errorf("workers %q: already started", absFileName)
  34. }
  35. workersRequestChans.Store(absFileName, make(chan *http.Request))
  36. shutdownWG.Add(nbWorkers)
  37. workersReadyWG.Add(nbWorkers)
  38. var (
  39. m sync.RWMutex
  40. errs []error
  41. )
  42. l := getLogger()
  43. for i := 0; i < nbWorkers; i++ {
  44. go func() {
  45. defer shutdownWG.Done()
  46. for {
  47. // Create main dummy request
  48. r, err := http.NewRequest(http.MethodGet, filepath.Base(absFileName), nil)
  49. if err != nil {
  50. panic(err)
  51. }
  52. r, err = NewRequestWithContext(
  53. r,
  54. WithRequestDocumentRoot(filepath.Dir(absFileName), false),
  55. WithRequestEnv(env),
  56. )
  57. if err != nil {
  58. panic(err)
  59. }
  60. l.Debug("starting", zap.String("worker", absFileName))
  61. if err := ServeHTTP(nil, r); err != nil {
  62. panic(err)
  63. }
  64. fc := r.Context().Value(contextKey).(*FrankenPHPContext)
  65. if fc.currentWorkerRequest != 0 {
  66. // Terminate the pending HTTP request handled by the worker
  67. maybeCloseContext(fc.currentWorkerRequest.Value().(*http.Request).Context().Value(contextKey).(*FrankenPHPContext))
  68. fc.currentWorkerRequest.Delete()
  69. fc.currentWorkerRequest = 0
  70. }
  71. // TODO: make the max restart configurable
  72. if _, ok := workersRequestChans.Load(absFileName); ok {
  73. workersReadyWG.Add(1)
  74. if fc.exitStatus == 0 {
  75. l.Info("restarting", zap.String("worker", absFileName))
  76. } else {
  77. l.Error("unexpected termination, restarting", zap.String("worker", absFileName), zap.Int("exit_status", int(fc.exitStatus)))
  78. }
  79. } else {
  80. break
  81. }
  82. }
  83. // TODO: check if the termination is expected
  84. l.Debug("terminated", zap.String("worker", absFileName))
  85. }()
  86. }
  87. workersReadyWG.Wait()
  88. m.Lock()
  89. defer m.Unlock()
  90. if len(errs) == 0 {
  91. return nil
  92. }
  93. return fmt.Errorf("workers %q: error while starting: %w", fileName, errors.Join(errs...))
  94. }
  95. func stopWorkers() {
  96. workersRequestChans.Range(func(k, v any) bool {
  97. workersRequestChans.Delete(k)
  98. return true
  99. })
  100. }
  101. //export go_frankenphp_worker_ready
  102. func go_frankenphp_worker_ready() {
  103. workersReadyWG.Done()
  104. }
  105. //export go_frankenphp_worker_handle_request_start
  106. func go_frankenphp_worker_handle_request_start(mrh C.uintptr_t) C.uintptr_t {
  107. mainRequest := cgo.Handle(mrh).Value().(*http.Request)
  108. fc := mainRequest.Context().Value(contextKey).(*FrankenPHPContext)
  109. v, ok := workersRequestChans.Load(fc.scriptFilename)
  110. if !ok {
  111. // Probably shutting down
  112. return 0
  113. }
  114. rc := v.(chan *http.Request)
  115. l := getLogger()
  116. l.Debug("waiting for request", zap.String("worker", fc.scriptFilename))
  117. var r *http.Request
  118. select {
  119. case <-done:
  120. l.Debug("shutting down", zap.String("worker", fc.scriptFilename))
  121. return 0
  122. case r = <-rc:
  123. }
  124. fc.currentWorkerRequest = cgo.NewHandle(r)
  125. l.Debug("request handling started", zap.String("worker", fc.scriptFilename), zap.String("url", r.RequestURI))
  126. if err := updateServerContext(r, false, mrh); err != nil {
  127. // Unexpected error
  128. l.Debug("unexpected error", zap.String("worker", fc.scriptFilename), zap.String("url", r.RequestURI), zap.Error(err))
  129. return 0
  130. }
  131. return C.uintptr_t(fc.currentWorkerRequest)
  132. }
  133. //export go_frankenphp_finish_request
  134. func go_frankenphp_finish_request(mrh, rh C.uintptr_t, deleteHandle bool) {
  135. rHandle := cgo.Handle(rh)
  136. r := rHandle.Value().(*http.Request)
  137. fc := r.Context().Value(contextKey).(*FrankenPHPContext)
  138. if deleteHandle {
  139. rHandle.Delete()
  140. cgo.Handle(mrh).Value().(*http.Request).Context().Value(contextKey).(*FrankenPHPContext).currentWorkerRequest = 0
  141. }
  142. maybeCloseContext(fc)
  143. var fields []zap.Field
  144. if mrh == 0 {
  145. fields = append(fields, zap.String("worker", fc.scriptFilename), zap.String("url", r.RequestURI))
  146. } else {
  147. fields = append(fields, zap.String("url", r.RequestURI))
  148. }
  149. fc.logger.Debug("request handling finished", fields...)
  150. }