线程调度轮询/调度调度

Question

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <semaphore.h> 
#define NUM_THREADS  4
#define COUNT_LIMIT 13
int     done = 0;
int     count = 0;
int     quantum = 2;
int     thread_ids[4] = {0,1,2,3};
int     thread_runtime[4] = {0,5,4,7};
pthread_mutex_t count_mutex;
pthread_cond_t count_threshold_cv;

void * inc_count(void * arg);


static sem_t count_sem;


 int quit = 0;

///////// Inc_Count////////////////
void *inc_count(void *t) 
{
   long my_id = (long)t;
   int i;
   sem_wait(&count_sem);   /////////////CRIT SECTION//////////////////////////////////      
   printf("run_thread = %d\n",my_id);
   printf("%d \n",thread_runtime[my_id]);
   for( i=0; i < thread_runtime[my_id];i++)
      {
         printf("runtime= %d\n",thread_runtime[my_id]);
         pthread_mutex_lock(&count_mutex); 
         count++;
         if (count == COUNT_LIMIT) {
            pthread_cond_signal(&count_threshold_cv);
            printf("inc_count(): thread %ld, count = %d  Threshold reached.\n", my_id,
            count);
         }
      printf("inc_count(): thread %ld, count = %d, unlocking mutex\n",my_id, count);
      pthread_mutex_unlock(&count_mutex);
      sleep(1) ;
      }//End For
   sem_post(&count_sem); // Next Thread Enters Crit Section
   pthread_exit(NULL);
}

/////////// Count_Watch ////////////////
void *watch_count(void *t) 
{
  long my_id = (long)t;
  printf("Starting watch_count(): thread %ld\n", my_id);

  pthread_mutex_lock(&count_mutex);
  if (count<COUNT_LIMIT) {
     pthread_cond_wait(&count_threshold_cv, &count_mutex);
     printf("watch_count(): thread %ld Condition signal received.\n", my_id);
     printf("watch_count(): thread %ld count now = %d.\n", my_id, count);
     }
  pthread_mutex_unlock(&count_mutex);
  pthread_exit(NULL);
}


////////////////// Main ////////////////
int main (int argc, char *argv[])
{
   int i;
   long t1=0, t2=1, t3=2, t4=3;
   pthread_t threads[4];
   pthread_attr_t attr;
   sem_init(&count_sem, 0, 1);
   /* Initialize mutex and condition variable objects */
   pthread_mutex_init(&count_mutex, NULL);
   pthread_cond_init (&count_threshold_cv, NULL);
   /* For portability, explicitly create threads in a joinable state */
   pthread_attr_init(&attr);
   pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
   pthread_create(&threads[0], &attr, watch_count, (void *)t1);
   pthread_create(&threads[1], &attr, inc_count, (void *)t2);
   pthread_create(&threads[2], &attr, inc_count, (void *)t3);
   pthread_create(&threads[3], &attr, inc_count, (void *)t4);
   /* Wait for all threads to complete */
   for (i=0; i<NUM_THREADS; i++) {
      pthread_join(threads[i], NULL);
   }
   printf ("Main(): Waited on %d  threads. Done.\n", NUM_THREADS);
   /* Clean up and exit */
   pthread_attr_destroy(&attr);
   pthread_mutex_destroy(&count_mutex);
   pthread_cond_destroy(&count_threshold_cv);
   pthread_exit(NULL);
  }

我正在尝试学习线程调度，有很多技术编码我都不知道。理论上我确实知道它应该如何工作，但在开始编写代码时遇到了麻烦……

我知道，至少我认为，这个程序不是实时的，它不应该是实时的。我需要创建一个调度器分派程序来控制线程的运行顺序...RR、FCFS、SJF等。

现在我没有调度员。我拥有的是信号量/互斥锁来控制线程。

此代码确实运行FCFS...我一直在尝试使用信号量来创建一个RR。但是有很多麻烦。我相信创建一个调度程序会更容易，但我不知道如何创建。

我需要帮助，我不是在寻找答案，只是在寻找方向..一些示例代码将有助于理解更多内容。

我遇到的问题是，似乎没有以这种方式同步线程的好方法。除非有一种方法可以为每个线程创建一个唯一的信号量。这就是为什么我需要一些帮助或指导来创建一个dispatcher函数。

谢谢。

Answer 1

首先，操作系统是系统中唯一可以实际调度线程运行的实体。较新的Linux内核中最常见的调度器是静态优先级的FCFS和RR，以及SCHED_OTHER调度器，现在由完全公平的调度器实现。

您似乎混淆了“操作系统级调度”和“应用程序级调度”的概念。前者对您的应用程序及其语义一无所知。后者必须使用诸如信号量、队列等工具来实现。

实现一组以FCFS方式执行的线程的一种方法是创建一个FIFO队列，用互斥锁保护它，并在这个队列中放置令牌，允许线程知道什么时候轮到它们运行。

线程的伪代码为：

while (1)
   lock_mutex()
   next = pop_queue()
   if (next == me)
      do_my_work()
      unlock_mutex()
      break
   unlock_muteX()

请注意，此示例不应按原样使用。它需要消费者和生产者以及其他消费者之间的谨慎协调。

Answer 2

您正在实现的代码根本就不是一个调度程序。对于schedule()函数，它告诉您下一个任务是什么，每个任务有两个基本状态(ready和run)，如果任务处于就绪模式，则可以根据任务的优先级将其置于run模式。

当计时中断发生时，上下文切换将保存所有寄存器，因此当dispatcher再次调用中断的任务时，您可以在代码的该点返回。以下是MIPS体系结构的一个示例。

int schedule(void){
    int lub_CurrentTskOrder = 0;
    int lub_IndexTask = 0;
    int lub_NextTsk2Run = 0;
    int lub_x = 0;
    int lub_y = 0;

    /* SEARCH THE RUNNING CURRENT TASK */
    for(lub_x = 0; lub_x < rub_InitializedTasks; lub_x++) {
        if(rs_SchedTask[lub_x].ub_task_status == tskRun) {
            rs_SchedTask[lub_x].ub_task_status = tskReady;
            lub_IndexTask = lub_x;
        }
    }

    if(rub_FirstDispatch == FALSE) {
        rs_SchedTask[lub_IndexTask].ub_task_status = tskRun;
        return lub_IndexTask;
    }

    for(lub_x = 0; lub_x < MAX_TASKS_NUMBER; lub_x++) {
        if(rs_SchedTask[lub_IndexTask].ub_ExecutionOrder == (rub_InitializedTasks - 1))
            lub_CurrentTskOrder = 0;
        else
            lub_CurrentTskOrder = rs_SchedTask[lub_IndexTask].ub_ExecutionOrder + 1;

        /* SEARCH THE NEXT EXECUTION TASK IN READY MODE */
        for(lub_y = 0; lub_y < MAX_TASKS_NUMBER; lub_y++) {
            if((rs_SchedTask[lub_y].ub_task_status == tskReady) && (rs_SchedTask[lub_y].ub_ExecutionOrder == lub_CurrentTskOrder)) {
                rs_SchedTask[lub_y].ub_task_status = tskRun;
                return lub_y;
            }
            else if((rs_SchedTask[lub_y].ub_task_status != tskReady) && (rs_SchedTask[lub_y].ub_ExecutionOrder == lub_CurrentTskOrder)) {
                lub_IndexTask = lub_y;
                break;
            }
        }
    }

    return(0);
}

/* INITIALIZE A STACK ON THE HEAP FOR AN SPECIFIC TASK */

S_PID sched_alloc(T_UBYTE lub_TaskNumber, S_TASK *lps_TaskStart)
{
    T_ULONG lul_x = 0;
    S_PID ls_pid_t;

    if((rub_InitializedTasks <= MAX_TASKS_NUMBER) && (rs_SchedTask[rub_InitializedTasks].ub_StackInit != TRUE)) {
        rs_SchedTask[rub_InitializedTasks].ub_ExecutionOrder = lub_TaskNumber;
        rs_SchedTask[rub_InitializedTasks].pfu_Entry = lps_TaskStart->pfu_Entry;                                                           // (1) STORE THE TASK ADDRESS TO INITIALIZE THE SCHEDULER
        rs_SchedTask[rub_InitializedTasks].pul_TaskFrame = malloc(((TASK_CONTEXT_STACK + lps_TaskStart->ul_StackSize) * 4) + 1);       // (2) CREATES A FRAME ON THE HEAP FOR THE CURRENT TASK
        ls_pid_t.pul_TaskFrame = rs_SchedTask[rub_InitializedTasks].pul_TaskFrame;

        for(lul_x = 0; lul_x < (TASK_CONTEXT_STACK + lps_TaskStart->ul_StackSize); lul_x++)     {                                              // (3) CLEAN ALL THE REGISTER SPACES ON THE CURRENT STACK
            rs_SchedTask[rub_InitializedTasks].pul_TaskFrame++;
            *rs_SchedTask[rub_InitializedTasks].pul_TaskFrame = 0x00;
        }
        rs_SchedTask[rub_InitializedTasks].pul_TaskFrame -= (TASK_CONTEXT_STACK + lps_TaskStart->ul_StackSize);                            // (4) RETURN TO THE STACK POSITION
        rs_SchedTask[rub_InitializedTasks].pul_TaskFrame++;
        *rs_SchedTask[rub_InitializedTasks].pul_TaskFrame = (T_ULONG)lps_TaskStart->pfu_Entry;                                         // (5) SAVE THE RETURN ADDRESS FOR THE TASK
        rs_SchedTask[rub_InitializedTasks].pul_TaskFrame--;
        rs_SchedTask[rub_InitializedTasks].pul_TaskStack = rs_SchedTask[rub_InitializedTasks].pul_TaskFrame + TASK_CONTEXT_STACK;  // (6) SET STACK FRAME ROOM FOR THE TASK
        rs_SchedTask[rub_InitializedTasks].pul_TaskStack += 0x54;                                                                      // (7) MAKE ROOM FOR REGISTERS ON STACK FRAME
        *rs_SchedTask[rub_InitializedTasks].pul_TaskFrame = (T_ULONG)rs_SchedTask[rub_InitializedTasks].pul_TaskStack;
        rps_CurrentTask = &rs_SchedTask[rub_InitializedTasks];
        asm_dispatcher_save_stack_pointer_on_stack;
        rs_SchedTask[rub_InitializedTasks].ub_StackInit = TRUE;                                                                        // (8) INDICATES THAT THE TASK IS ALLREADY INITIALIZED ON STACK
        rs_SchedTask[rub_InitializedTasks].psb_TaskName = lps_TaskStart->psb_TaskName;
        rub_InitializedTasks++;                                                                                                        // (9) THIS IS THE INITIALIZED TASKS COUNTER FOR PUBLIC USE
        ls_pid_t.pfu_Entry = lps_TaskStart->pfu_Entry;                                                                                 // (10) SAVE THE PID VALUES
        ls_pid_t.psb_TaskName = lps_TaskStart->psb_TaskName;
        return(ls_pid_t);                                                                                                          // (11) RETURN PID
    }
}

第一个函数返回下一个要运行的任务，因此您可以调用dispatcher来进行上下文切换。滴答中断服务例程像调用下一段代码一样调用汇编器调度程序。

void __interrupt(TICK) TICK_ISR(void)
{
    atomic_cstart();
    mips_r3000_reset_interval_timer();
    rub_CurrentTask = schedule();

    // save current context
    __asm volatile                                 \
    (                                              \
        "and $k1,$k1,$zero                                     \n\t" \
        "or $k1,$k1,rps_CurrentTask                            \n\t" \
        "lw $k1,0x0000($k1)       ;Pointer to Task Structure   \n\t" \
        "lw $k1,0x0004($k1)       ;Pointer to Task Stack Frame \n\t" \
        "sw $sp,0x0000($k1)       ;Save GPR $sp                \n\t" \
        "and $sp,$sp,$zero                                     \n\t" \
        "or $sp,$sp,$k1           ;Pointer to Task Stack Frame \n\t" \
        "mfc0 $k1,$ER                                          \n\t" \
        "sw $k1,0x0004($sp)       ;Save Return Address         \n\t" \
        "sw $s0,0x0008($sp)       ;Save GPR $s0                \n\t" \
        "sw $s1,0x000c($sp)       ;Save GPR $s1                \n\t" \
        "sw $s2,0x0010($sp)       ;Save GPR $s2                \n\t" \
        "sw $s3,0x0014($sp)       ;Save GPR $s3                \n\t" \
        "sw $s4,0x0018($sp)       ;Save GPR $s4                \n\t" \
        "sw $s5,0x001c($sp)       ;Save GPR $s5                \n\t" \
        "sw $s6,0x0020($sp)       ;Save GPR $s6                \n\t" \
        "sw $s7,0x0024($sp)       ;Save GPR $s7                \n\t" \
        "sw $s8,0x0028($sp)       ;Save GPR $fp                \n\t" \
        "lw $k1,0x0000($sp)                                    \n\t" \
        "and $sp,$sp,$zero                                     \n\t" \
        "or $sp,$sp,$k1           ;Restore GPR $sp             \n\t" \
        "and $k1,$k1,$zero                                     \n\t" \
    )

    rps_CurrentTask = &rs_SchedTask[rub_CurrentTask];

    // restore current context

    __asm volatile                                 \
    (                                              \
        "and $sp,$sp,$zero                                     \n\t" \
        "or $sp,$sp,rps_CurrentTask                            \n\t" \
        "lw $sp,0x0000($sp)       ;Pointer to Task Structure   \n\t" \
        "lw $sp,0x0004($sp)       ;Pointer to Task Stack Frame \n\t" \
        "lw $k1,0x0004($sp)                                    \n\t" \
        "mtc0 $k1,$ER             ;Restore Return Address      \n\t" \
        "lw $s0,0x0008($sp)       ;Restore GPR $s0             \n\t" \
        "lw $s1,0x000c($sp)       ;Restore GPR $s1             \n\t" \
        "lw $s2,0x0010($sp)       ;Restore GPR $s2             \n\t" \
        "lw $s3,0x0014($sp)       ;Restore GPR $s3             \n\t" \
        "lw $s4,0x0018($sp)       ;Restore GPR $s4             \n\t" \
        "lw $s5,0x001c($sp)       ;Restore GPR $s5             \n\t" \
        "lw $s6,0x0020($sp)       ;Restore GPR $s6             \n\t" \
        "lw $s7,0x0024($sp)       ;Restore GPR $s7             \n\t" \
        "lw $s8,0x0028($sp)       ;Restore GPR $fp             \n\t" \
        "lw $k1,0x0000($sp)                                    \n\t" \
        "and $sp,$sp,$zero                                     \n\t" \
        "or $sp,$sp,$k1           ;Restore GPR $sp             \n\t" \
        "and $k1,$k1,$zero                                     \n\t" \
    )
    atomic_cend();
}

这些操作是原子的，因为您必须禁用中断来保存和恢复dispatcher汇编例程的当前上下文。