diff --git a/src/frontend/apiserver/backend/clib.go b/src/frontend/apiserver/backend/clib.go index f5bcd50..51a848d 100644 --- a/src/frontend/apiserver/backend/clib.go +++ b/src/frontend/apiserver/backend/clib.go @@ -9,6 +9,7 @@ package backend #include "client.h" #include "netio.h" #include "helper.h" +#include "idgen.h" #include "controller.capnp.h" // backend_ctx_t wraps xcecli_ctx_t for use as the C handler's arg. @@ -54,6 +55,7 @@ import ( "apiserver/models" logger "apiserver/utils" "fmt" + "sync" "unsafe" "go.uber.org/zap" @@ -62,6 +64,10 @@ import ( // gCtx holds the C client context for the lifetime of the process. var gCtx C.backend_ctx_t +// gIDCtx is the C idgen context used to generate job IDs. +var gIDCtx C.idgen_ctx_t +var idCtxOnce sync.Once + // gHandler receives push notifications from the backend. // Must be set via StartNotificationListener before RegisterToBackend. var gHandler NotificationHandler @@ -113,6 +119,30 @@ func StartNotificationListener(handler NotificationHandler) error { return nil } +// initIDGen initialises the C idgen context. Called once via sync.Once. +func initIDGen() { + C.idgen_ctx_init(&gIDCtx) +} + +// GenerateJobID calls the C idgen() to produce a unique job ID. +// clusterID and apiserverID identify the originating instance; +// arraySize=1 for normal jobs, >1 for array jobs. +func GenerateJobID(clusterID, apiserverID int, arraySize int) (uint64, error) { + idCtxOnce.Do(initIDGen) + var jobID C.uint64_t + rc := C.idgen(&gIDCtx, C.size_t(clusterID), C.size_t(apiserverID), + C.size_t(arraySize), &jobID) + if rc < 0 { + return 0, fmt.Errorf("idgen failed: %d", int(rc)) + } + return uint64(jobID), nil +} + +// Cleanup releases C resources. Call at process shutdown. +func Cleanup() { + C.idgen_ctx_fini(&gIDCtx) +} + // go_push_handler is called by the C library on every incoming request. // It must be exported so the C trampoline can call it from a C thread. // @@ -204,12 +234,15 @@ func handleConfigPush(cfgdata *C.xce_configdata_t) C.int { } // SubmitJob sends a job submission to the backend via xcecli_jobsubmit. +// arraySize=1 for normal jobs, >1 for array jobs. // The store is updated only when the backend pushes the resulting OnJobCreated // notification — not immediately here. -func SubmitJob(job *models.JobInfo) error { +func SubmitJob(job *models.JobInfo, arraySize int) error { js, sd, si := modelJobToSubmit(job) defer freeJobSubmit(js, sd, si) + js.array_size = C.uint64_t(arraySize) + var outJobID C.uint64_t rc := C.xcecli_jobsubmit(gCtx.ctx, js, &outJobID) if rc < 0 { @@ -222,28 +255,6 @@ func SubmitJob(job *models.JobInfo) error { return nil } -// SubmitArrayJobs submits an array job via xcecli_jobsubmit with array_size > 1. -func SubmitArrayJobs(arrayjob *models.ArrayJobInfo, jobs []models.JobInfo) error { - if len(jobs) == 0 { - return fmt.Errorf("no jobs in array") - } - - // Use the first job as the template; set array_size to the total count. - js, sd, si := modelJobToSubmit(&jobs[0]) - defer freeJobSubmit(js, sd, si) - js.array_size = C.uint64_t(len(jobs)) - - var outJobID C.uint64_t - rc := C.xcecli_jobsubmit(gCtx.ctx, js, &outJobID) - if rc < 0 { - return fmt.Errorf("xcecli_jobsubmit (array) failed: %d", int(rc)) - } - if outJobID != 0 { - arrayjob.JobID = int64(outJobID) - } - return nil -} - // ControlJob sends a job control action to the backend via xcecli_jobctrl. // action must be one of the models.JOB_* constants; content is an optional // signal name (e.g. "TERM") for kill operations. diff --git a/src/frontend/apiserver/backend/stub.go b/src/frontend/apiserver/backend/stub.go index f38e516..f17cade 100644 --- a/src/frontend/apiserver/backend/stub.go +++ b/src/frontend/apiserver/backend/stub.go @@ -15,11 +15,13 @@ func RegisterToBackend(config *BackendConfig) error { return errNoCGO } // StartNotificationListener stub — real implementation in clib.go (requires CGO). func StartNotificationListener(handler NotificationHandler) error { return errNoCGO } -// SubmitJob stub — real implementation in clib.go (requires CGO). -func SubmitJob(job *models.JobInfo) error { return errNoCGO } +// GenerateJobID stub — real implementation in clib.go (requires CGO). +func GenerateJobID(clusterID, apiserverID int, arraySize int) (uint64, error) { + return 0, errNoCGO +} -// SubmitArrayJobs stub — real implementation in clib.go (requires CGO). -func SubmitArrayJobs(arrayjob *models.ArrayJobInfo, jobs []models.JobInfo) error { return errNoCGO } +// SubmitJob stub — real implementation in clib.go (requires CGO). +func SubmitJob(job *models.JobInfo, arraySize int) error { return errNoCGO } // ControlJob stub — real implementation in clib.go (requires CGO). func ControlJob(jobid uint64, action int, content string) error { return errNoCGO } diff --git a/src/frontend/apiserver/controllers/jobs.go b/src/frontend/apiserver/controllers/jobs.go index 4de5a37..a5f11ae 100644 --- a/src/frontend/apiserver/controllers/jobs.go +++ b/src/frontend/apiserver/controllers/jobs.go @@ -27,7 +27,8 @@ import ( const ( //TODO: Need assign clusterID globally. - clusterID = 1 + clusterID = 1 + apiserverID = 0 ) var ( @@ -36,10 +37,8 @@ var ( "resume": "CONT", "stop": "STOP", } - jobID_Worker = utils.NewWorker(clusterID) realCopyLocalFiles []string fakeCopyRemoteFile []string - indices []int ) type JobController struct{} @@ -1183,16 +1182,17 @@ func (j JobController) RestartJobByID(ctx *gin.Context) { } } - jobIDArray, err := generateJobIDs(indices) + jobID, err := backend.GenerateJobID(clusterID, apiserverID, 1) if err != nil { - ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) + logger.Error("Failed to generate job ID for restart", zap.Error(err)) + ReturnError(ctx, common.ERROR_CODE, "failed to generate job ID: "+err.Error(), nil) return } // Use a concrete type for job newJob := models.JobInfo{ - Unique_JobID: jobIDArray[0].Unique_JobID, - JobID: jobIDArray[0].Unique_JobID, + Unique_JobID: int64(jobID), + JobID: int64(jobID), ArrayIdx: 0, SubmitUser: restarter, JobStatus: models.JOBSTAT["PEND"], @@ -1341,10 +1341,9 @@ func (j JobController) RestartJobByID(ctx *gin.Context) { if v, ok := oldJob["login_shell"].(string); ok { newJob.LoginShell = v } - logger.Info("Generated job IDs", zap.Any("jobIDArray", jobIDArray), zap.Any("indices", len(indices))) + logger.Info("Generated job ID for restart", zap.Int64("jobID", newJob.JobID)) - // sequential job - err = services.SubmitJob(&newJob) + err = services.SubmitJob(&newJob, 1) if err != nil { logger.Error("Failed to submit job for restart process.", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, common.ERROR_WRITE_DATA_MSG+": "+err.Error(), nil) @@ -1385,17 +1384,14 @@ func (j JobController) CreateJob(ctx *gin.Context) { var req models.JobSubmitRequest - indices = []int{} if err := utils.BindJSONStrict(ctx, &req); err != nil { logger.Error("Invalid request parameters", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } - // save the original job name before parsing - arrayJobName := req.JobName // precheck the request parameter before generating the job ID - err = preCheckJobRequest(&req) + parsedIndices, err := preCheckJobRequest(&req) if err != nil { logger.Error("Failed to submit job request", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) @@ -1419,20 +1415,23 @@ func (j JobController) CreateJob(ctx *gin.Context) { return } } - // generate job IDs - jobIDArray, err := generateJobIDs(indices) + + // Determine array size: 1 for normal jobs, >1 for array jobs + arraySize := 1 + if len(parsedIndices) > 1 { + arraySize = len(parsedIndices) + } + + // Generate job ID via C idgen (single call for both normal and array jobs) + jobID, err := backend.GenerateJobID(clusterID, apiserverID, arraySize) if err != nil { - ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) + logger.Error("Failed to generate job ID", zap.Error(err)) + ReturnError(ctx, common.ERROR_CODE, "failed to generate job ID: "+err.Error(), nil) return } - logger.Info("Generated job IDs", zap.Any("jobIDArray", jobIDArray), zap.Any("indices", len(indices))) - - // get the primary jobID - var primaryJobID int64 - if len(jobIDArray) > 0 { - primaryJobID = jobIDArray[0].Unique_JobID - } + primaryJobID := int64(jobID) + logger.Info("Generated job ID", zap.Int64("jobID", primaryJobID), zap.Int("arraySize", arraySize)) defaultValues := map[string]string{ "job_id": fmt.Sprintf("%d", primaryJobID), @@ -1451,96 +1450,68 @@ func (j JobController) CreateJob(ctx *gin.Context) { return } - jobs := make([]models.JobInfo, len(jobIDArray)) - for i, jobID := range jobIDArray { - logger.Debug("Create new job info for jobID", zap.Int64("jobID", jobID.Unique_JobID)) - - // Create new job info - jobs[i] = models.JobInfo{ - Unique_JobID: jobID.Unique_JobID, - JobID: primaryJobID, - ArrayIdx: jobID.ArrayIdx, - //SubmitTime: time.Now().Unix(), // backend will fill the submit time; - SubmitUser: submitUser, - JobStatus: models.JOBSTAT["PEND"], // new job is pend. - JobName: req.JobName, - Command: req.Command, - QueueName: req.QueueName, //TODO: need set to DEFAULT_QUEUE - ProjectName: req.ProjectName, //TODO: need set "default" if not specify? - UserGroup: req.UserGroup, - LicenseProject: req.LicenseProject, - JobDescription: req.JobDescription, - ResReq: req.Resreq, - RsvID: req.RsvID, - SLA: req.SLA, - MaxNumProcessors: req.MaxNumProcessors, - NumProcessors: req.NumProcessors, - AskedHosts: req.AskedHosts, - LaunchType: req.LaunchType, - //BeginTime: req.BeginTime, - //TermTime: req.TermTime, - BeginTime: models.CustomTime{Time: req.BeginTime.Time}, - TermTime: models.CustomTime{Time: req.TermTime.Time}, - JobPriority: req.JobPriority, - UserPriority: req.JobPriority, - DependCond: req.DependCond, - CWD: req.CWD, - FromHost: ctx.ClientIP(), //use ip now, need use hostname ? - PreExecCmd: req.PreExecCmd, - PostExecCmd: req.PostExecCmd, - RequeueErrorValues: req.RequeueErrorValues, - RunTimeEstimation: req.RunTimeEstimation, - MigThreshold: req.MigThreshold, - ChkpntPeriod: req.ChkpntPeriod, - ChkpntDir: req.ChkpntDir, - OutDir: req.OutDir, - InFile: req.InFile, // transfer file into json with absolute path - ErrFile: req.ErrFile, - OutFile: req.OutFile, - LoginShell: req.LoginShell, - ExtendData: models.Jsonb{String: req.ExtendData}, - } - - err = postJobRequest(&jobs[i]) - if err != nil { - logger.Error("Failed to handle postJobRequest ", zap.Error(err)) - ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) - return - } - logger.Debug("jobinfo before submit.", zap.Any("job", jobs[i])) + // Build the single JobInfo to submit + job := models.JobInfo{ + Unique_JobID: primaryJobID, + JobID: primaryJobID, + //SubmitTime: time.Now().Unix(), // backend will fill the submit time; + SubmitUser: submitUser, + JobStatus: models.JOBSTAT["PEND"], // new job is pend. + JobName: req.JobName, + Command: req.Command, + QueueName: req.QueueName, //TODO: need set to DEFAULT_QUEUE + ProjectName: req.ProjectName, //TODO: need set "default" if not specify? + UserGroup: req.UserGroup, + LicenseProject: req.LicenseProject, + JobDescription: req.JobDescription, + ResReq: req.Resreq, + RsvID: req.RsvID, + SLA: req.SLA, + MaxNumProcessors: req.MaxNumProcessors, + NumProcessors: req.NumProcessors, + AskedHosts: req.AskedHosts, + LaunchType: req.LaunchType, + //BeginTime: req.BeginTime, + //TermTime: req.TermTime, + BeginTime: models.CustomTime{Time: req.BeginTime.Time}, + TermTime: models.CustomTime{Time: req.TermTime.Time}, + JobPriority: req.JobPriority, + UserPriority: req.JobPriority, + DependCond: req.DependCond, + CWD: req.CWD, + FromHost: ctx.ClientIP(), //use ip now, need use hostname ? + PreExecCmd: req.PreExecCmd, + PostExecCmd: req.PostExecCmd, + RequeueErrorValues: req.RequeueErrorValues, + RunTimeEstimation: req.RunTimeEstimation, + MigThreshold: req.MigThreshold, + ChkpntPeriod: req.ChkpntPeriod, + ChkpntDir: req.ChkpntDir, + OutDir: req.OutDir, + InFile: req.InFile, // transfer file into json with absolute path + ErrFile: req.ErrFile, + OutFile: req.OutFile, + LoginShell: req.LoginShell, + ExtendData: models.Jsonb{String: req.ExtendData}, } - if len(jobs) == 1 { - // sequential job - err = services.SubmitJob(&jobs[0]) - if err != nil { - logger.Error("Failed to submit new sequential job.", zap.Error(err)) - ReturnError(ctx, common.ERROR_CODE, common.ERROR_WRITE_DATA_MSG+": "+err.Error(), nil) - return - } - } else { - // array job - arrayJob := models.ArrayJobInfo{ - JobID: primaryJobID, - ArrayJobName: arrayJobName, - SubmitUser: submitUser, - QueueName: req.QueueName, - UserGroup: req.UserGroup, - JobDescription: req.JobDescription, - JobName: req.JobName, - ProjectName: req.ProjectName, - LicenseProject: req.LicenseProject, - SLA: req.SLA, - } - err = services.SubmitArrayJobs(arrayJob, jobs) - if err != nil { - logger.Error("Failed to write new array jobs.", zap.Error(err)) - ReturnError(ctx, common.ERROR_CODE, common.ERROR_WRITE_DATA_MSG+": "+err.Error(), nil) - return - } + err = postJobRequest(&job) + if err != nil { + logger.Error("Failed to handle postJobRequest ", zap.Error(err)) + ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) + return + } + logger.Debug("jobinfo before submit.", zap.Any("job", job)) + + // Submit job to backend (single call handles both normal and array) + err = services.SubmitJob(&job, arraySize) + if err != nil { + logger.Error("Failed to submit job.", zap.Error(err)) + ReturnError(ctx, common.ERROR_CODE, common.ERROR_WRITE_DATA_MSG+": "+err.Error(), nil) + return } - logger.Info("Job submission successfully from ", zap.String("client_ip", ctx.ClientIP()), zap.Any("job", jobs[0])) + logger.Info("Job submission successfully from ", zap.String("client_ip", ctx.ClientIP()), zap.Int64("jobID", primaryJobID)) response := models.JobSubmitResponse{ Message: fmt.Sprintf("Job <%d> is submitted to default queue <%s>.", primaryJobID, req.QueueName), @@ -1548,37 +1519,6 @@ func (j JobController) CreateJob(ctx *gin.Context) { QueueName: req.QueueName, } ReturnSuccess(ctx, common.CREATED, common.SUCCESS_MSG, response) - -} - -func generateJobIDs(indices []int) ([]models.JobArrayID, error) { - logger.Debug("Enter generateJobIDs() function", zap.Any("indices", indices)) - // initialize jobIDPairs - jobIDPairs := make([]models.JobArrayID, 0, len(indices)) - - // if indices is empty, generate the primary jobid - if len(indices) == 0 { - jobID, err := jobID_Worker.NextID() - if err != nil { - logger.Error("Failed to generate job id.", zap.Error(err)) - return nil, fmt.Errorf("failed to generate job id: %w", err) - } - jobIDPairs = append(jobIDPairs, models.JobArrayID{ArrayIdx: 0, Unique_JobID: jobID}) - return jobIDPairs, nil - } - - // generate the other job IDs based on indices - for _, index := range indices { - jobID, err := jobID_Worker.NextID() - if err != nil { - logger.Error("Failed to generate job id.", zap.Int("index", index), zap.Error(err)) - return nil, fmt.Errorf("failed to generate job id: %w", err) - } - jobIDPairs = append(jobIDPairs, models.JobArrayID{ArrayIdx: index, Unique_JobID: jobID}) - } - - logger.Debug("Generated jobIDPairs", zap.Any("jobIDPairs", jobIDPairs)) - return jobIDPairs, nil } func preCheckJobRestartRequest(req *models.JobRestartRequest) error { @@ -1589,7 +1529,7 @@ func preCheckJobRestartRequest(req *models.JobRestartRequest) error { return nil } -func preCheckJobRequest(req *models.JobSubmitRequest) error { +func preCheckJobRequest(req *models.JobSubmitRequest) ([]int, error) { logger.Debug("Enter preCheckJobRequest") if req.QueueName == "" { req.QueueName = "normal" @@ -1598,33 +1538,32 @@ func preCheckJobRequest(req *models.JobSubmitRequest) error { queueInfo, ok := store.Queues.Get(req.QueueName) if !ok { logger.Error("Invalid queue name specified", zap.String("queue_name", req.QueueName)) - return fmt.Errorf("invalid queue name: %s", req.QueueName) + return nil, fmt.Errorf("invalid queue name: %s", req.QueueName) } queue_statusStr := formatQueueStatus(int64(queueInfo.QueueStatus)) if strings.HasPrefix(strings.ToLower(queue_statusStr), "closed") { logger.Error("Queue is closed", zap.String("queue_name", req.QueueName)) - return fmt.Errorf("cannot submit job to queue %s: queue is closed", req.QueueName) + return nil, fmt.Errorf("cannot submit job to queue %s: queue is closed", req.QueueName) } } if req.JobName != "" { - arrayName, parsedIndices, slotLimit, err := createArrayName(req.JobName) + arrayName, idx, slotLimit, err := createArrayName(req.JobName) if err != nil { logger.Error("job name format is invalid", zap.Error(err)) - return errors.New("job array list format is invalid: " + err.Error()) + return nil, errors.New("job array list format is invalid: " + err.Error()) } if arrayName != "" { if err = isValidArrayName(arrayName); err != nil { logger.Error(err.Error()) - return errors.New("job array list format is invalid: " + err.Error()) + return nil, errors.New("job array list format is invalid: " + err.Error()) } } - indices = parsedIndices req.JobName = arrayName logger.Debug("preCheckJobRequest parsed job name is: ", zap.String("jobName", req.JobName), - zap.Any("indices", indices), zap.Any("slotLimit", slotLimit)) - return nil + zap.Any("indices", idx), zap.Any("slotLimit", slotLimit)) + return idx, nil } else { if req.Command != "" { if len(req.Command) > 4094 { @@ -1637,24 +1576,24 @@ func preCheckJobRequest(req *models.JobSubmitRequest) error { if req.JobDescription != "" { if len(req.JobDescription) > 4094 { - return errors.New("job description length exceeds the limit of 4094 characters") + return nil, errors.New("job description length exceeds the limit of 4094 characters") } } // Validate AskedHosts using the new function if err := validateAskedHosts(req.AskedHosts); err != nil { - return err + return nil, err } if req.ProjectName != "" { if len(req.ProjectName) > 511 { - return errors.New("project name length exceeds the limit of 511 characters") + return nil, errors.New("project name length exceeds the limit of 511 characters") } } if req.Resreq != "" { if len(req.Resreq) > 4094 { - return errors.New("resreq length exceeds the limit of 4094 characters") + return nil, errors.New("resreq length exceeds the limit of 4094 characters") } } @@ -1695,16 +1634,16 @@ func preCheckJobRequest(req *models.JobSubmitRequest) error { // handle environment variables if req.Env != nil { if len(req.Env) > 4094 { - return errors.New("environment variable length exceeds the limit of 4094 characters") + return nil, errors.New("environment variable length exceeds the limit of 4094 characters") } for _, value := range req.Env { parseEnvs := strings.SplitN(value, "=", 2) if len(parseEnvs) != 2 { - return errors.New("environment variable should use key=value format") + return nil, errors.New("environment variable should use key=value format") } key := parseEnvs[0] if strings.TrimSpace(key) == "" { - return errors.New("environment variable key cannot be empty") + return nil, errors.New("environment variable key cannot be empty") } } } @@ -1717,7 +1656,7 @@ func preCheckJobRequest(req *models.JobSubmitRequest) error { for _, transferFileStr := range req.TransferFiles { localFile, operator, remoteFile, err := parseTransferFiles(transferFileStr) if err != nil { - return err + return nil, err } /* based on different operator, check the local file and remote file @@ -1728,7 +1667,7 @@ func preCheckJobRequest(req *models.JobSubmitRequest) error { */ if operator == ">" || operator == "<>" || operator == "><" { if localFile == "" { - return errors.New("local file is required for operator " + operator) + return nil, errors.New("local file is required for operator " + operator) } // Append to realCopyLocalFiles for later processing realCopyLocalFiles = append(realCopyLocalFiles, localFile) @@ -1736,7 +1675,7 @@ func preCheckJobRequest(req *models.JobSubmitRequest) error { if operator == "<" || operator == "<<" { if remoteFile == "" { - return errors.New("remote file is required for operator " + operator) + return nil, errors.New("remote file is required for operator " + operator) } fakeCopyRemoteFile = append(fakeCopyRemoteFile, remoteFile) } @@ -1751,7 +1690,7 @@ func preCheckJobRequest(req *models.JobSubmitRequest) error { // Convert inFileData to JSON and assign to req.InFile extendDataJSON, err := json.Marshal(extendData) if err != nil { - return fmt.Errorf("failed to marshal InFile JSON: %v", err) + return nil, fmt.Errorf("failed to marshal InFile JSON: %v", err) } req.ExtendData = string(extendDataJSON) } @@ -1763,7 +1702,7 @@ func preCheckJobRequest(req *models.JobSubmitRequest) error { minTasks, maxTasks, err := checkNumProcessors(minTasks, maxTasks) if err != nil { logger.Error("Failed to parse NumProcessors", zap.Any("numProcessors", req.NumProcessors), zap.Error(err)) - return fmt.Errorf("invalid NumProcessors specification: %w", err) + return nil, fmt.Errorf("invalid NumProcessors specification: %w", err) } req.NumProcessors = minTasks req.MaxNumProcessors = maxTasks @@ -1772,22 +1711,22 @@ func preCheckJobRequest(req *models.JobSubmitRequest) error { } if req.MigThreshold < 0 || req.MigThreshold > 2147483647 { - return errors.New("Incorrect migration threshold value. Specify a number greater than or equal to zero.") + return nil, errors.New("Incorrect migration threshold value. Specify a number greater than or equal to zero.") } if !utils.ContainsField(models.ValidInteractiveValues, strconv.Itoa(req.Interactive)) { - return errors.New("Interactive field must be 0, 1, or 2") + return nil, errors.New("Interactive field must be 0, 1, or 2") } if req.LaunchType != "" { launchType := strings.ToLower(req.LaunchType) if !utils.ContainsField(models.ValidLaunchTypes, launchType) { - return errors.New("Launch_type must be openmpi or mpich") + return nil, errors.New("Launch_type must be openmpi or mpich") } req.LaunchType = launchType } - return nil + return nil, nil } // It returns minTasks, maxTasks, and an error if parsing fails or constraints are violated. diff --git a/src/frontend/apiserver/services/job_service.go b/src/frontend/apiserver/services/job_service.go index c7957ad..c853f2c 100644 --- a/src/frontend/apiserver/services/job_service.go +++ b/src/frontend/apiserver/services/job_service.go @@ -5,12 +5,7 @@ import ( "apiserver/models" ) -// SubmitJob sends a job to the backend. The store is updated via notification callback. -func SubmitJob(job *models.JobInfo) error { - return backend.SubmitJob(job) -} - -// SubmitArrayJobs sends an array job to the backend. -func SubmitArrayJobs(arrayjob models.ArrayJobInfo, joblist []models.JobInfo) error { - return backend.SubmitArrayJobs(&arrayjob, joblist) +// SubmitJob sends a job to the backend. arraySize=1 for normal, >1 for array. +func SubmitJob(job *models.JobInfo, arraySize int) error { + return backend.SubmitJob(job, arraySize) } diff --git a/src/frontend/apiserver/utils/jobidgen.go b/src/frontend/apiserver/utils/jobidgen.go deleted file mode 100644 index 8e65a8d..0000000 --- a/src/frontend/apiserver/utils/jobidgen.go +++ /dev/null @@ -1,80 +0,0 @@ -package utils - -import ( - "errors" - "sync" - "time" -) - -/** - * 31 bit timestamp(s) | | 5 bit clusterID | 10 bit sequenceBits - */ -const ( - clusterIDBits = uint64(5) // 5 bits clusterID - sequenceBits = uint64(10) - - maxClusterID = int64(-1) ^ (int64(-1) << clusterIDBits) - maxSequence = int64(-1) ^ (int64(-1) << sequenceBits) - - timeLeft = uint8(15) // timeLeft = clusterIDBits + sequenceBits // - clusterLeft = uint8(10) - - twepoch = int64(1735228800) // Dec 27 00:00:00 2024 -) - -type Worker struct { - mu sync.Mutex - LastStamp int64 // last timestamp - //WorkerID int64 // - ClusterID int64 // cluster id - Sequence int64 -} - -// under distributed env, we need assign uniform id to each cluster: max 32 cluster. -func NewWorker(clusterID int) *Worker { - return &Worker{ - //WorkerID: workerID, - LastStamp: 0, - Sequence: 0, - ClusterID: int64(clusterID), - } -} - -func (w *Worker) getSeconds() int64 { - //return time.Now().UnixNano() / 1e9 // get timestamp in second. - return time.Now().Unix() // / 1e9 // get timestamp in second. -} - -func (w *Worker) NextID() (int64, error) { - w.mu.Lock() - defer w.mu.Unlock() - - return w.nextID() -} - -func (w *Worker) nextID() (int64, error) { - timeStamp := w.getSeconds() - if timeStamp < w.LastStamp { - return 0, errors.New("time is moving backwards,waiting until") - } - - if w.LastStamp == timeStamp { - w.Sequence = (w.Sequence + 1) & maxSequence - - if w.Sequence == 0 { - for timeStamp <= w.LastStamp { - timeStamp = w.getSeconds() - } - } - } else { - w.Sequence = 0 - } - - w.LastStamp = timeStamp - - id := ((timeStamp - twepoch) << timeLeft) | - (w.ClusterID << clusterLeft) | - w.Sequence - - return int64(id), nil -}