void
minithread_yield() {
  //put current thread at end of runnable
  semaphore_P(runnable_q_lock);
  current_thread->priority = 0;
  current_thread->rem_quanta = 1;

  if (multilevel_queue_enqueue(runnable_q,
      current_thread->priority,current_thread) == 0) {
    runnable_count++;
  }
  semaphore_V(runnable_q_lock);
  //call scheduler here
  scheduler();
}

/**
 * Force unmount on all filesystems. This function should be used only
 * when halting the system. Waits for all VFS operations to complete
 * but does not wait for all files to be closed. After this function
 * is called the VFS and the whole operating system can no longer be
 * used.
 */
void vfs_deinit(void)
{
  fs_t *fs;
  int row;

  semaphore_P(vfs_op_sem);
  vfs_usable = 0;

  kprintf("VFS: Entering forceful unmount of all filesystems.\n");
  if (vfs_ops > 0) {
    kprintf("VFS: Delaying force unmount until the pending %d "
            "operations are done.\n", vfs_ops);
    semaphore_V(vfs_op_sem);
    semaphore_P(vfs_unmount_sem);
    semaphore_P(vfs_op_sem);
    KERNEL_ASSERT(vfs_ops == 0);
    kprintf("VFS: Continuing forceful unmount.\n");
  }

  semaphore_P(vfs_table.sem);
  semaphore_P(openfile_table.sem);

  for (row = 0; row < CONFIG_MAX_FILESYSTEMS; row++) {
    fs = vfs_table.filesystems[row].filesystem;
    if (fs != NULL) {
      kprintf("VFS: Forcefully unmounting volume [%s]\n",
              vfs_table.filesystems[row].mountpoint);
      fs->unmount(fs);
      vfs_table.filesystems[row].filesystem = NULL;
    }
  }

  semaphore_V(openfile_table.sem);
  semaphore_V(vfs_table.sem);
  semaphore_V(vfs_op_sem);
}

/* Deletes all the elements in the cleanup_queue and then get the next
 * RUNNABLE minithread, if it exists. Otherwise, it will continuously poll the
 * queue and wait until another thread is made RUNNABLE */
int reaper_queue_cleanup(arg_t arg) {
  //delete all zombie minithreads in cleanup_queue
  minithread_t* temp;
  while (1) {
    semaphore_P(reaper_sema); //waits until something is ready to be deleted
    if (queue_length(schedule_data->cleanup_queue) > 0) { //double checks size
      interrupt_level_t old_level = set_interrupt_level(DISABLED); 
      queue_dequeue(schedule_data->cleanup_queue, (void **) &temp);
      free(temp);
      set_interrupt_level(old_level);
    }
  }
  // will never return
  return 0;
}

static int
employee(int* arg) {
    phone_t new_phone = NULL;
    int id = employee_id++;

    while (1) {
        /* Check if there is any customer in need of a phone */
        semaphore_P(customer_sem);

        /* Wait if buffer full */
        semaphore_P(empty_sem);
        new_phone = phone_create();
        if (NULL != new_phone) {
            queue_append(phone_queue, new_phone);
            printf("Employee %d unpacked phone %d\n", id, new_phone->serial_num);
        }
        /* Signal a phone is ready */
        semaphore_V(full_sem);

        /* minithread_yield(); */
    }

    return 0;
}

/* produce all integers from 2 to max */
int source(int* arg) {
    channel_t* c = (channel_t *) arg;
    int i;

    for (i=2; i<=max; i++) {
        c->value = i;
        semaphore_V(c->consume);
        semaphore_P(c->produce);
    }

    c->value = -1;
    semaphore_V(c->consume);

    return 0;
}

static void minisocket_free (minisocket_t * socket) {
  network_interrupt_arg_t *packet = NULL;
  while (queue_dequeue(socket->data, (void **)&packet) != -1) {
    free(packet);
  }
  queue_free(socket->data);
  socket->data = NULL;
  semaphore_destroy(socket->data_ready);
  semaphore_destroy(socket->wait_for_ack);
  semaphore_destroy(socket->send_receive_mutex);
  semaphore_P(ports_mutex);
  ports[socket->local_port] = NULL;
  semaphore_V(ports_mutex);
  free(socket);
}

// returns new task id
uint32_t sched_request_task(task_type type, std::shared_ptr<void> data)
{
	uint32_t task_id;
	std::unique_ptr<new_task_req> request(new new_task_req);

	request->tcp = data;
	request->type = type;
	request->task_id = task_counter++;
	task_id = request->task_id;
	
	semaphore_P(sched_new_task_lock, 1);
	new_task_queue.push_back(std::move(request));
	semaphore_V(sched_new_task_lock, 1);

	return task_id;
}

/**
 * Start a new operation on VFS. Operation is defined to be any such
 * sequence of actions (a VFS function call) that may touch some
 * filesystem.
 *
 * @return VFS_OK if operation can continue, error (negative) if
 * operation must be cancelled.
 */
static int vfs_start_op()
{
    int ret = VFS_OK;

    semaphore_P(vfs_op_sem);

    if (vfs_usable) {
        vfs_ops++;
    } else {
        ret = VFS_UNUSABLE;
    }

    semaphore_V(vfs_op_sem);

    return ret;
}

/* Creates an unbound port for listening. Multiple requests to create the same
 * unbound port should return the same miniport reference. It is the responsibility
 * of the programmer to make sure he does not destroy unbound miniports while they
 * are still in use by other threads -- this would result in undefined behavior.
 * Unbound ports must range from 0 to 32767. If the programmer specifies a port number
 * outside this range, it is considered an error.
 */
miniport_t
miniport_create_unbound(int port_number) {
  miniport_t new_port;

  if (port_number < 0 || port_number >= BOUND_PORT_START) {
    // user error
    return NULL;
  }
  if (miniport_array[port_number]) {
    return miniport_array[port_number];
  }
  semaphore_P(unbound_ports_lock);
  new_port = (miniport_t)malloc(sizeof(struct miniport)); 
  if (new_port == NULL) {
    semaphore_V(unbound_ports_lock);
    return NULL;
  }
  new_port->p_type = UNBOUND_PORT;
  new_port->p_num = port_number;
  new_port->u.unbound.port_pkt_q = queue_new();
  if (!new_port->u.unbound.port_pkt_q) {
    free(new_port);
    semaphore_V(unbound_ports_lock);
    return NULL;
  }
  new_port->u.unbound.port_pkt_available_sem = semaphore_create();
  if (!new_port->u.unbound.port_pkt_available_sem) {
    queue_free(new_port->u.unbound.port_pkt_q);
    free(new_port);
    semaphore_V(unbound_ports_lock);
    return NULL;
  } 
  new_port->u.unbound.q_lock = semaphore_create();
  if (!new_port->u.unbound.q_lock) {
    queue_free(new_port->u.unbound.port_pkt_q);
    semaphore_destroy(new_port->u.unbound.port_pkt_available_sem);
    free(new_port);
    semaphore_V(unbound_ports_lock);
    return NULL;
  }
  semaphore_initialize(new_port->u.unbound.port_pkt_available_sem,0);
  semaphore_initialize(new_port->u.unbound.q_lock,1);
  miniport_array[port_number] = new_port;
  semaphore_V(unbound_ports_lock);
  return new_port;
   
}

static void ctl_program(int ch, int val)
{
        if(ch >= MAX_TK_MIDI_CHANNELS)
		return;
	if (!ctl.trace_playing)
		return;
	if (ch < 0 || ch >= MAX_TK_MIDI_CHANNELS) return;
	if(channel[ch].special_sample)
	    val = channel[ch].special_sample;
	else
	    val += progbase;

	semaphore_P(semid);
	Panel->channel[ch].program = val;
	Panel->c_flags[ch] |= FLAG_PROG;
	semaphore_V(semid);
}

/**
 * End a VFS operation.
 */
static void vfs_end_op()
{
    semaphore_P(vfs_op_sem);

    vfs_ops--;

    KERNEL_ASSERT(vfs_ops >= 0);

    /* Wake up pending unmount if VFS is now idle. */
    if (!vfs_usable && (vfs_ops == 0))
        semaphore_V(vfs_unmount_sem);

    if (!vfs_usable && (vfs_ops > 0))
        kprintf("VFS: %d operations still pending\n", vfs_ops);

    semaphore_V(vfs_op_sem);
}

int vfs_seek(openfile_t file, int seek_position)
{
    openfile_entry_t *openfile;

    if (vfs_start_op() != VFS_OK)
        return VFS_UNUSABLE;

    KERNEL_ASSERT(seek_position >= 0);
    semaphore_P(openfile_table.sem);

    openfile = vfs_verify_open(file);
    openfile->seek_position = seek_position;

    semaphore_V(openfile_table.sem);

    vfs_end_op();
    return VFS_OK;
}

/* sends a miniroute packet, automatically discovering the path if necessary.
 * See description in the .h file.
 */
int
miniroute_send_pkt(network_address_t dest_address, int hdr_len, char* hdr,
                   int data_len, char* data)
{
    int total_len = hdr_len + data_len;
    int sent_len = 0;
    struct routing_header routing_hdr;
    char *total_data = malloc(total_len);
    miniroute_path_t path = NULL;

    /* Find route if it is not in cache */
    semaphore_P(discovery_mutex);
    miniroute_cache_get_by_addr(route_cache,  dest_address, (void**)&path);

    if (NULL == path || path->exp_time < ticks) {
#if MINIROUTE_CACHE_DEBUG == 1
        printf("Address not found, discovering path.\n");
#endif
        path = miniroute_discover_path(dest_address);
    }

    semaphore_V(discovery_mutex);
    if (NULL != path && NULL != (total_data = malloc(total_len))) {
        /* Pack data and miniroute header */
        memcpy(total_data, hdr, hdr_len);
        memcpy(total_data + hdr_len, data, data_len);

        miniroute_pack_data_hdr(&routing_hdr, path);
        sent_len = network_send_pkt(path->hop[1], MINIROUTE_HDRSIZE,
                                    (char*)&routing_hdr, total_len, total_data);
    }
#if MINIROUTE_DEBUG == 1
    printf("Network packet sent.\n");
#endif
#if MINIROUTE_CACHE_DEBUG == 1
    printf("Sending data with header: \n");
    miniroute_print_hdr(&routing_hdr);
#endif

    if (sent_len < hdr_len)
        return -1;
    else
        return sent_len - hdr_len;
}

/* Creates a bound port for use in sending packets. The two parameters, addr and
 * remote_unbound_port_number together specify the remote's listening endpoint.
 * This function should assign bound port numbers incrementally between the range
 * 32768 to 65535. Port numbers should not be reused even if they have been destroyed,
 * unless an overflow occurs (ie. going over the 65535 limit) in which case you should
 * wrap around to 32768 again, incrementally assigning port numbers that are not
 * currently in use.
 */
miniport_t
miniport_create_bound(network_address_t addr, int remote_unbound_port_number)
{
    int num;
    semaphore_P(port_mutex);
    num = miniport_get_boundedport_num();
    if (num < MIN_BOUNDED || num > MAX_BOUNDED) {
        semaphore_V(port_mutex);
        return NULL;
    }
    if ((port[num] = malloc(sizeof(struct miniport))) != NULL) {
        port[num]->type = BOUNDED;
        port[num]->num = num;
        network_address_copy(addr, port[num]->bound.addr);
        port[num]->bound.remote = remote_unbound_port_number;
    }
    semaphore_V(port_mutex);
    return port[num];
}

std::string sched_get_running_tasks()
{
	std::stringstream ss;

	semaphore_P(sched_lock, 1);
	for (int i = 0; i < CORE_COUNT; i++)
	{
		if (sched_active_task(i))
		{
			ss << i << "| ";
			ss << running_tasks[i]->task_id << ' ';
			ss << task_state_names[running_tasks[i]->state] << ' ';
			ss << task_type_names[running_tasks[i]->type] << '\n';
		}
	}
	semaphore_V(sched_lock, 1);

	return ss.str();
}

void minisocket_destroy(minisocket_t sock, minisocket_error* error){
  network_interrupt_arg_t* pkt;

  semaphore_P(server_lock);
  sock_array[sock->src_port] = NULL;
  semaphore_V(server_lock);
  *error = SOCKET_NOERROR;
  //free all queued packets. interrupts not disabled
  //because socket in array set to null, network
  //handler cannot access queue
  while (queue_dequeue(sock->pkt_q,(void**)&pkt) != -1){
    free(pkt);
  }
  queue_free(sock->pkt_q);
  semaphore_destroy(sock->pkt_ready_sem);
  semaphore_destroy(sock->ack_ready_sem);
  semaphore_destroy(sock->sock_lock);
  free(sock);
}

std::string sched_get_cores_info()
{
	std::stringstream ss;

	semaphore_P(sched_lock, 1);
	for (int i = 0; i < CORE_COUNT; i++)
	{
		if (sched_active_task(i))
		{
			ss << "core " << i << " occupied" << '\n';
		}
		else {
			ss << "core " << i << " free" << '\n';
		}
	}
	semaphore_V(sched_lock, 1);

	return ss.str();
}

int main()
{
    struct exchange
    {
        char buf[BUFSIZ+80];
        int seq;
    }shm;
    int shmid;
    unsigned char *retval;
    int producer, consumer, i;
    char readbuf[BUFSIZ];

    //创建信号量consumer
    consumer = open_semaphore_set(ftok("consumer", 0), 1);

    //初始化信号量consumer的值为1
    init_a_semaphore(consumer, 0, 1);

    //创建信号量producer
    producer = open_semaphore_set(ftok("producer", 0), 1);

    //初始化信号量producer的值为1
    init_a_semaphore(producer, 0, 1);

    //创建共享存储区用于存放生产者产生的数据
    shmid = shmget(ftok("share", 0),sizeof(struct exchange), 0666|IPC_CREAT);
    retval = shmat(shmid, (unsigned char *)0, 0);

    //生产者与消费者同步
    for (i = 0; ; i++)
    {
        printf("enter some text:");
        fgets(readbuf, BUFSIZ, stdin);
        semaphore_P(consumer);
        sprintf(retval, "messge %2d form producer %d is \" %s \" ", i, getpid(), readbuf);
        semaphore_V(producer);
        if (strncmp(readbuf, "end", 3) == 0)
            break;
    }
    exit(0);

    return 0;
}

/*
 * Destroys a miniport and frees up its resources. If the miniport was in use at
 * the time it was destroyed, subsequent behavior is undefined.
 */
void
miniport_destroy(miniport_t miniport)
{
    semaphore_P(port_mutex);
    if (NULL == miniport) {
        semaphore_V(port_mutex);
        return;
    }
    if (UNBOUNDED == miniport->type) {
        queue_free(miniport->unbound.data);
        semaphore_destroy(miniport->unbound.ready);
    }
    if (BOUNDED == miniport->type) {
        --bound_count;
    }
    port[miniport->num] = NULL;
    free(miniport);
    semaphore_V(port_mutex);
}

int vfs_mount(fs_t *fs, char *name)
{
    int i;
    int row;

    KERNEL_ASSERT(name != NULL && name[0] != '\0');

    if (vfs_start_op() != VFS_OK)
        return VFS_UNUSABLE;

    semaphore_P(vfs_table.sem);

    for (i = 0; i < CONFIG_MAX_FILESYSTEMS; i++) {
        if (vfs_table.filesystems[i].filesystem == NULL)
            break;
    }

    row = i;

    if(row >= CONFIG_MAX_FILESYSTEMS) {
        semaphore_V(vfs_table.sem);
        kprintf("VFS: Warning, maximum mount count exceeded, mount failed.\n");
        vfs_end_op();
        return VFS_LIMIT;
    }

    for (i = 0; i < CONFIG_MAX_FILESYSTEMS; i++) {
        if(stringcmp(vfs_table.filesystems[i].mountpoint, name) == 0) {
            semaphore_V(vfs_table.sem);
            kprintf("VFS: Warning, attempt to mount 2 filesystems "
                    "with same name\n");
            vfs_end_op();
            return VFS_ERROR;
        }
    }

    stringcopy(vfs_table.filesystems[row].mountpoint, name, VFS_NAME_LENGTH);
    vfs_table.filesystems[row].filesystem = fs;

    semaphore_V(vfs_table.sem);
    vfs_end_op();
    return VFS_OK;
}