学步园

在开源的数据库中，支持并行建索引的，只有ingres，但它并不是并行建一个索引，而是对同一个表，同时并行建多个索引，

下面主要分析它如何协调主子线程通信部分，使用版本是 ingres-10.1.0-114-gpl；

并行建索引函数为：src/back/dmf/dmu/dm2upind.c        STATUS dm2u_pindex( DM2U_INDEX_CB   *index_cbs)

主子线程通信函数是 src/back/dmf/dmu/dm2uputi.c       

 DB_STATUS dm2u_pload_table(DM2U_MXCB   *mxcbs, i4  *rec_cnt, DB_ERROR  *dberr)

ingres是多线程，所以建索引的主线程和各个子线程是通过内存直接交换信息，没有通过共享内存；它由多个页面组成

/*<br /> ** Name: exch_hdr<br /> **<br /> ** Description:<br /> ** This structure is used to describe the exchange block. It describes<br /> ** how many exchange block is there and its beginning address. Also<br /> ** the header keeps track of the status of the parent thread should<br /> ** a wakeup is needed.<br /> **<br /> ** History:<br /> ** 10-apr-1998 (nanpr01)<br /> ** Created for Parallel Index Build.<br /> ** 这就是交换信息的头结构<br /> */<br /> struct _EXCH_BUF_HDR<br /> {<br /> i4 ehdr_noofbufs; /* No of exchange buffers */<br /> i4 ehdr_rec_capacity; /* capacity of each exchange buffer */<br /> i4 ehdr_size; /* Size of each exchange buffer */<br /> EXCH_BUF *ehdr_exch_buffers; //指向交换页面的首地址<br /> i4 ehdr_parent_status; /* if this is non-zero, parent is<br /> ** waiting on that buffer */<br /> CS_SEMAPHORE ehdr_cwait_mutex; /* mutex for parent to synchro-<br /> ** nize children execution<br /> */<br /> i4 ehdr_cwait_count; /* Contains wait count */<br /> i4 ehdr_error_flag;<br /> #define PARENT_ERROR 0x00000001L<br /> #define CHILD_ERROR 0x00000002L<br /> };</p> <p>/*}<br /> ** Name: exch_buf<br /> **<br /> ** Description:<br /> ** This structure is used to describe the individual exchange buffers.<br /> ** This contains its size, the number in its list and a count field<br /> ** which all readers increment when that buffer is being read and<br /> ** decremented when done.<br /> **<br /> ** History:<br /> ** 10-apr-1998 (nanpr01)<br /> ** Created for Parallel Index Build.<br /> ** 05-may-2000 (stial01)<br /> ** Added rec_cnt to prototype for dm2u_pload_table<br /> ** 每个交换页面的头结构<br /> */<br /> struct _EXCH_BUF {<br /> CS_SEMAPHORE exch_buf_mutex; /* Mutex to protect sim access */<br /> i4 exch_bufno; /* buffer number for the exch buf */<br /> char *exch_buffer; /* actual exchange buffer 指向实际的页面 */<br /> i4 exch_noofrec; /* no of record holding */<br /> CS_SEMAPHORE exch_cnt_mutex; /* mutex to protect count */<br /> i4 exch_usr_cnt; /* no of readers to read from<br /> ** this buffer<br /> */<br /> char exch_visit[512/BITSPERBYTE];/* Who visited this buffer<br /> ** b115572: increase max concurrent<br /> ** indexes to 512<br /> */<br /> };<br /> //exch_buffer的大小，不是常见的固定页面大小8K，而是<br /> //(sizeof(DM_TID) + 每个元组大小) * no_of_records;</p> <p>/*}<br /> ** Name: child_cb<br /> **<br /> ** Description:<br /> ** This structure is used to describe the control block passed to the<br /> ** child. In this control block, mxcb address and the exchange buffer<br /> ** header is passed<br /> **<br /> ** History:<br /> ** 10-apr-1998 (nanpr01)<br /> ** Created for Parallel Index Build.<br /> ** 每个子线程从主线程得到的数据结构<br /> */<br /> struct _CHILD_CB {<br /> i4 ccb_noofchild; /* No of fac. thread created */<br /> EXCH_BUF_HDR *ccb_exch_hdr; /* Exchange Buffer header */<br /> DM2U_MXCB *ccb_mxcb; /* mxcb for each create ind */<br /> i4 ccb_childno; /* This contains the children<br /> ** number<br /> */<br /> i4 ccb_thread_id; /* child thread id */<br /> DB_STATUS ccb_status; /* child status */<br /> DB_ERROR ccb_error; /* child error array */<br /> };</p> <p>

DB_STATUS dm2u_pload_table(DM2U_MXCB   *mxcbs, i4  *rec_cnt, DB_ERROR  *dberr)

/*<br /> ** calculate the size of the buffers to allocate for parent-child<br /> ** communication and exchange buffers<br /> ** 交换用的各个头指针和页面的大小<br /> */<br /> size_of_buffer = (sizeof(DM_TID) + reclen) * no_of_records;<br /> exch_tot_size = sizeof(EXCH_BUF_HDR) +<br /> no_of_buffers * (size_of_buffer + sizeof(EXCH_BUF)) +<br /> (sizeof(CHILD_CB) * no_of_ind);</p> <p> /* Create the factotum threads for parallel index creation */<br /> //是直接创建线程，创建后把新建的线程链入一个链表，链表还没仔细看<br /> //ftc.ftc_thread_entry = build_pindex;<br /> //ftc.ftc_thread_exit = build_pindex_exit;<br /> //上面两个就是给子线程的执行函数</p> <p> status = create_index_threads(mxcbs, ccb, exch_hdr,<br /> (EXCH_BUF *)exch_buf, no_of_ind, dberr);</p> <p>//主线程向交换页面不停的放入数据，放满一个页面后，子线程会取走，不满的话，子线程看不到，除非是最后一个页面<br /> while ( status == E_DB_OK && !end_of_file &&<br /> !(m->mx_flags & MX_ONLINE_INDEX_BUILD) )<br /> {<br /> /* get a record from the base table */<br /> local_status = dm2r_get(m->mx_rcb, &tid, DM2R_GETNEXT,<br /> dummy_record, dberr);<br /> /* check to see if user interrupt occurred. */<br /> if ( local_status == E_DB_OK && *(m->mx_rcb->rcb_uiptr) )<br /> {<br /> /* If XCB, check via SCB */<br /> if ( m->mx_rcb->rcb_xcb_ptr )<br /> {<br /> dmxCheckForInterrupt(m->mx_rcb->rcb_xcb_ptr, &error);<br /> if ( error )<br /> SETDBERR(dberr, 0, error);<br /> }<br /> else if (*(m->mx_rcb->rcb_uiptr) & RCB_USER_INTR)<br /> SETDBERR(dberr, 0, E_DM0065_USER_INTR);<br /> /* check to see if force abort occurred. */<br /> else if (*(m->mx_rcb->rcb_uiptr) & RCB_FORCE_ABORT)<br /> SETDBERR(dberr, 0, E_DM010C_TRAN_ABORTED);</p> <p> if ( dberr->err_code )<br /> local_status = E_DB_ERROR;<br /> }</p> <p> if (local_status == E_DB_OK)<br /> *rec_cnt = *rec_cnt + 1;</p> <p> /* put record in exchange buffer */<br /> status = put_record(exch_hdr, &cur_buf, &tid, dummy_record,<br /> reclen, no_of_ind, &end_of_file,<br /> local_status, dberr);<br /> }</p> <p> /*<br /> ** We have finished loading the data to the child threads now<br /> ** and child threads can go concurrent<br /> ** We have to wait for all the children to finish.<br /> ** If one of them finishes with error, we have to tell others<br /> ** to stop at the earliest opportunities and return error<br /> ** when everyone is done.<br /> */<br /> wait_for_index_finish(exch_hdr);

/*{<br /> ** Name: build_pindex<br /> **<br /> ** Description:<br /> ** This routine gets executed by each child creating index. This reads<br /> ** the exchange buffer and passes it to sorter. At end, this routine<br /> ** calls the load routine to build the index in parallel.<br /> **<br /> ** Inputs:<br /> ** Outputs:<br /> ** Returns:<br /> ** E_DB_OK<br /> ** E_DB_ERROR<br /> ** Exceptions:<br /> ** none<br /> **<br /> ** Side Effects:<br /> ** none<br /> **<br /> ** History:<br /> ** 10-apr-98 (nanpr01)<br /> ** Created.<br /> ** 25-Jan-2002 (jenjo02)<br /> ** Must hold exch_buf_mutex while checking parent_status,<br /> ** and parent_status must be cleared before CSresume-ing<br /> ** parent to ensure that only one child resumes the<br /> ** parent.<br /> ** Set/return error status directly in child's CHILD_CB<br /> ** where it can be viewed by the parent.<br /> ** 15-Jan-2007 (horda03) Bug 117471<br /> ** If a child thread is waiting for its parent to populate a buffer<br /> ** (or other children to catch up deschedule the child to give them<br /> ** a chance to do the work.<br /> ** This is esential if using Internal threads.<br /> */<br /> //子线程用于创建索引的主函数<br /> STATUS build_pindex(SCF_FTX *ftx)<br /> {<br /> CHILD_CB *ccb;<br /> EXCH_BUF *cur_buf;<br /> EXCH_BUF_HDR *exch_hdr;<br /> i4 allread = FALSE, noofrecord, reclen;<br /> DM_OBJECT *wloc_mask = NULL;<br /> DM2U_MXCB *m;<br /> DM2U_TPCB *tp;<br /> DM_TID tid;<br /> i4 local_err_code;<br /> i4 error;<br /> DB_ERROR local_dberr;</p> <p> ccb = (CHILD_CB *)ftx->ftx_data;<br /> exch_hdr = ccb->ccb_exch_hdr;<br /> m = ccb->ccb_mxcb;<br /> tp = m->mx_tpcb_next;</p> <p> /* round robin buffer for reading and that is why always start at 1 */<br /> cur_buf = exch_hdr->ehdr_exch_buffers;<br /> reclen = (exch_hdr->ehdr_size/<br /> (exch_hdr->ehdr_rec_capacity) - sizeof(DM_TID));</p> <p> if ( (ccb->ccb_status = init_child_thread(tp, &wloc_mask, &ccb->ccb_error)) )<br /> return(ccb->ccb_status);</p> <p> if (m->mx_flags & MX_ONLINE_INDEX_BUILD)<br /> return(ccb->ccb_status);</p> <p> /*<br /> ** This routine waits for parents to send data in exchange buffer<br /> ** until nomore data<br /> */<br /> //从主线程接收数据的主循环<br /> while ( ccb->ccb_status == E_DB_OK && !allread )<br /> {<br /> #ifdef xDEBUG<br /> TRdisplay("child %d Going to wait on buffer %d/n",<br /> ccb->ccb_childno, cur_buf->exch_bufno);<br /> #endif<br /> /* Wait to get a buffer using the shared mutex */<br /> CSp_semaphore(FALSE, &cur_buf->exch_buf_mutex);<br /> /* check for errors before reading data */<br /> if (exch_hdr->ehdr_error_flag)<br /> {<br /> ccb->ccb_status = E_DB_WARN;<br /> /* if parent is waiting on this buffer .. wake him up */<br /> if (exch_hdr->ehdr_parent_status == cur_buf->exch_bufno)<br /> {<br /> exch_hdr->ehdr_parent_status = 0;<br /> CSresume(ftx->ftx_thread_id);<br /> }<br /> CSv_semaphore(&cur_buf->exch_buf_mutex);<br /> break;<br /> }<br /> #ifdef xDEBUG<br /> TRdisplay("child %d Got buffer mutex for buffer %d/n",<br /> ccb->ccb_childno, cur_buf->exch_bufno);<br /> #endif</p> <p> /* Mutex the buffer */<br /> CSp_semaphore(TRUE, &cur_buf->exch_cnt_mutex);<br /> //当一个页面没写满时，可通过下面的判断，跳过该页面<br /> if (!cur_buf->exch_usr_cnt)<br /> {<br /> /* Child switched to this buffer before the parent filled it */<br /> #ifdef xDEBUG<br /> TRdisplay(" CHILD %d buffer %x %d has no users no_of_records %d/n",<br /> ccb->ccb_childno, cur_buf, cur_buf->exch_bufno,<br /> cur_buf->exch_noofrec );<br /> #endif<br /> /* if parent is waiting on this buffer .. wake him up */<br /> if (exch_hdr->ehdr_parent_status == cur_buf->exch_bufno)<br /> {<br /> exch_hdr->ehdr_parent_status = 0;<br /> CSresume(ftx->ftx_thread_id);<br /> }<br /> CSv_semaphore(&cur_buf->exch_cnt_mutex);<br /> CSv_semaphore(&cur_buf->exch_buf_mutex);</p> <p> /* Let the parent have a chance to run */<br /> CSswitch();<br /> continue;<br /> }</p> <p> /* Have we visited this buffer already */<br /> if (BTtest(ccb->ccb_childno, (char *)&cur_buf->exch_visit))<br /> {<br /> #ifdef xDEBUG<br /> TRdisplay("child %d already visited this buffer %d/n",<br /> ccb->ccb_childno, cur_buf->exch_bufno);<br /> #endif</p> <p> /* Wait for other children to catch up */<br /> CSv_semaphore(&cur_buf->exch_cnt_mutex);<br /> CSv_semaphore(&cur_buf->exch_buf_mutex);</p> <p> /* Let the other children have a chance to run */<br /> CSswitch();<br /> continue;<br /> }<br /> CSv_semaphore(&cur_buf->exch_cnt_mutex);<br /> //遇到非满页面，认为主线程已把所有数据发送完毕<br /> if (cur_buf->exch_noofrec != exch_hdr->ehdr_rec_capacity)<br /> {<br /> /* This is the last buffer to read */<br /> allread = TRUE;<br /> }<br /> #ifdef xDEBUG<br /> TRdisplay("child %d reading buffer %d end %d/n",<br /> ccb->ccb_childno, cur_buf->exch_bufno, allread);<br /> #endif<br /> if ( noofrecord = cur_buf->exch_noofrec )<br /> {<br /> i4 nth;</p> <p> while ( noofrecord > 0 && ccb->ccb_status == E_DB_OK )<br /> {<br /> nth = cur_buf->exch_noofrec - noofrecord;<br /> get_record(exch_hdr, cur_buf,<br /> tp, nth, reclen, &tid);<br /> ccb->ccb_status = load_sorter(tp, &tid, &ccb->ccb_error);<br /> noofrecord--;<br /> }</p> <p> CSp_semaphore(TRUE, &cur_buf->exch_cnt_mutex);<br /> cur_buf->exch_usr_cnt--;<br /> BTset(ccb->ccb_childno, (char *)&cur_buf->exch_visit);<br /> if (!cur_buf->exch_usr_cnt)<br /> {<br /> /* if parent is waiting on this buffer .. wake him up */<br /> if (exch_hdr->ehdr_parent_status == cur_buf->exch_bufno)<br /> {<br /> exch_hdr->ehdr_parent_status = 0;<br /> CSresume(ftx->ftx_thread_id);<br /> }<br /> }<br /> CSv_semaphore(&cur_buf->exch_cnt_mutex);<br /> }<br /> CSv_semaphore(&cur_buf->exch_buf_mutex);<br /> #ifdef xDEBUG<br /> TRdisplay("child %d finished reading this buffer %d/n",<br /> ccb->ccb_childno, cur_buf->exch_bufno);<br /> #endif<br /> if ( ccb->ccb_status == E_DB_OK )<br /> {<br /> if (allread)<br /> {<br /> ccb->ccb_status = dmse_input_end(tp->tpcb_srt, &ccb->ccb_error);<br /> }<br /> else if (cur_buf->exch_bufno == exch_hdr->ehdr_noofbufs)<br /> /* reset back to 1 */<br /> cur_buf = exch_hdr->ehdr_exch_buffers;<br /> else<br /> cur_buf++;<br /> }<br /> }<br /> //子线程数据接收完毕，开始建索引<br /> if ( ccb->ccb_status == E_DB_OK )<br /> ccb->ccb_status = load_index(tp, wloc_mask, &ccb->ccb_error);<br /> else<br /> {<br /> if ( dmse_end(tp->tpcb_srt, &local_dberr) )<br /> {<br /> uleFormat(&local_dberr, 0, (CL_ERR_DESC *)NULL, ULE_LOG, NULL,<br /> (char *)NULL, (i4)0, (i4 *)NULL, &local_err_code, 0);<br /> if (local_dberr.err_code > ccb->ccb_error.err_code)<br /> ccb->ccb_error = local_dberr;<br /> }<br /> tp->tpcb_srt = 0;<br /> }<br /> if (wloc_mask != NULL)<br /> dm0m_deallocate((DM_OBJECT **)&wloc_mask);<br /> return(ccb->ccb_status);<br /> }

在子线程从交换页面读数据时，是互斥的，就是说同一时刻只有一个子线程读一个交换页面，不存在并发读

而在主线程写的时候，是没加互斥的，

一个写，多个读的时候，读之间互斥，写没加互斥，没写满之前，读不到

 主线程放数据，子线程取数据的函数，没什么东西，主要是写满之后，取下一个页面的函数

 /*{<br /> ** Name: get_record<br /> **<br /> ** Description:<br /> ** This routine reads nth record from the exchange buffer. This routine<br /> ** does not make any attempt to check whether such a record exists.<br /> ** It is the responsibility of the caller to check this.<br /> **<br /> ** Inputs:<br /> ** Outputs:<br /> ** Returns:<br /> ** E_DB_OK<br /> ** Exceptions:<br /> ** none<br /> **<br /> ** Side Effects:<br /> ** none<br /> **<br /> ** History:<br /> ** 10-apr-98 (nanpr01)<br /> ** Created.<br /> */<br /> static STATUS<br /> get_record(<br /> EXCH_BUF_HDR *exch_buf_hdr,<br /> EXCH_BUF *cur_buf,<br /> DM2U_TPCB *tp,<br /> i4 nth,<br /> i4 reclen,<br /> DM_TID *tid<br /> )<br /> {<br /> char *p;<br /> i4 i;</p> <p> p = cur_buf->exch_buffer + (nth * (reclen + sizeof(DM_TID)));<br /> MEcopy(p, sizeof(DM_TID), tid);<br /> p += sizeof(DM_TID);<br /> MEcopy(p, reclen, tp->tpcb_crecord);</p> <p>#ifdef xDEBUG<br /> /* Now got it - so read it */<br /> TRdisplay("Child Record in buffer %d pos %d TID :",<br /> cur_buf->exch_bufno, nth);<br /> TRdisplay("%d/tRecord :", tid->tid_i4);<br /> for (i = 0; i < reclen; i++)<br /> {<br /> TRdisplay("%1c", tp->tpcb_crecord[i]);<br /> }<br /> TRdisplay("/nAddresses 0x%x 0x%x/n", p-sizeof(DM_TID), p);<br /> #endif<br /> return(E_DB_OK);<br /> }</p> <p>/*{<br /> ** Name: put_record<br /> **<br /> ** Description:<br /> ** This routine gets a record from the base table and puts it in the<br /> ** current exchange buffer. If the current exchange buffer is full,<br /> ** it gets the next exchage buffer by calling switch_buffer routine.<br /> **<br /> ** Inputs:<br /> ** Outputs:<br /> ** Returns:<br /> ** E_DB_OK<br /> ** E_DB_ERROR<br /> ** Exceptions:<br /> ** none<br /> **<br /> ** Side Effects:<br /> ** none<br /> **<br /> ** History:<br /> ** 10-apr-98 (nanpr01)<br /> ** Created.<br /> ** 05-may-2000 (stial01, gupsh01)<br /> ** If E_DM0055_NONEXT, and current buffer is full, the parent<br /> ** switches to the next buffer. Set exch_usr_cnt and exch_visit<br /> ** in the next buffer.<br /> */<br /> static STATUS<br /> put_record(<br /> EXCH_BUF_HDR *exch_buf_hdr,<br /> EXCH_BUF **cur_buf_ptr,<br /> DM_TID *tid,<br /> char *record,<br /> i4 reclen,<br /> i4 noofchild,<br /> i4 *end_of_file,<br /> i4 local_status,<br /> DB_ERROR *dberr)<br /> {<br /> char *start;<br /> i4 i;<br /> EXCH_BUF *cur_buf = *cur_buf_ptr;<br /> STATUS status; </p> <p> if (local_status == E_DB_OK)<br /> {<br /> if ((cur_buf->exch_noofrec + 1) > exch_buf_hdr->ehdr_rec_capacity)<br /> {<br /> status = switch_buffer(exch_buf_hdr, cur_buf_ptr, noofchild);<br /> if (status != E_DB_OK)<br /> return(E_DB_ERROR);<br /> cur_buf = *cur_buf_ptr;<br /> }</p> <p> start = cur_buf->exch_buffer +<br /> (cur_buf->exch_noofrec * (reclen + sizeof(DM_TID)));<br /> MEcopy((char*)tid, sizeof(DM_TID), start);<br /> MEcopy(record, reclen, start+sizeof(DM_TID));</p> <p>#ifdef xDEBUG<br /> TRdisplay("Parent Record in buffer %d pos %d TID: ",<br /> cur_buf->exch_bufno,<br /> cur_buf->exch_noofrec);<br /> TRdisplay("%d/t Record:", tid->tid_i4);<br /> for (i = 0; i < reclen; i++)<br /> {<br /> TRdisplay("%1c", record[i]);<br /> }<br /> TRdisplay("/n Addresses 0x%x 0x%x/n", start, start+sizeof(DM_TID));<br /> #endif</p> <p> (cur_buf->exch_noofrec)++;<br /> return(E_DB_OK);<br /> }<br /> else {<br /> //处理最后的未满页面<br /> if (local_status == E_DB_ERROR && dberr->err_code == E_DM0055_NONEXT)<br /> {<br /> /* if the current buffer is full, let children read it */<br /> if ((cur_buf->exch_noofrec + 1) > exch_buf_hdr->ehdr_rec_capacity)<br /> {<br /> status = switch_buffer(exch_buf_hdr, cur_buf_ptr, noofchild);<br /> if (status != E_DB_OK)<br /> return(E_DB_ERROR);<br /> cur_buf = *cur_buf_ptr;<br /> }</p> <p> /*<br /> ** Always set exch_usr_cnt and exch_visit when parent gets EOF<br /> ** so that the child can detect if it switches to a buffer<br /> ** before the parent.<br /> */<br /> cur_buf->exch_usr_cnt = noofchild;<br /> MEfill(sizeof(cur_buf->exch_visit), '/0', &cur_buf->exch_visit);</p> <p> CSv_semaphore(&cur_buf->exch_buf_mutex);<br /> /*<br /> ** End of table - Tell children to start load phase<br /> */<br /> CLRDBERR(dberr);<br /> *end_of_file = 1;<br /> return(E_DB_OK);<br /> }<br /> else {<br /> /* error occurred while reading - so bail out */</p> <p> /* Tell children that parent got error */<br /> exch_buf_hdr->ehdr_error_flag = PARENT_ERROR;<br /> CSv_semaphore(&cur_buf->exch_buf_mutex);</p> <p> return(E_DB_ERROR);<br /> }<br /> }<br /> }

 /*{<br /> ** Name: switch_buffer<br /> **<br /> ** Description:<br /> ** This routine checks if the caller has an error. If it doesnot<br /> ** it gets the next available buffer when available.<br /> **<br /> ** Inputs:<br /> ** Outputs:<br /> ** Returns:<br /> ** E_DB_OK<br /> ** E_DB_ERROR<br /> ** Exceptions:<br /> ** none<br /> **<br /> ** Side Effects:<br /> ** none<br /> **<br /> ** History:<br /> ** 10-apr-98 (nanpr01)<br /> ** Created.<br /> ** 25-Jan-2002 (jenjo02)<br /> ** Reorganized code to ensure that "parent_status" is<br /> ** set and reset while holding exch_buf_mutex.<br /> */<br /> static STATUS<br /> switch_buffer(EXCH_BUF_HDR *exch_buf_hdr, EXCH_BUF **cur_buf_ptr,<br /> i4 noofchild)<br /> {<br /> EXCH_BUF *next_buf;<br /> EXCH_BUF *cur_buf = *cur_buf_ptr;</p> <p> /* done with this buffer */<br /> //经过下面的两个操作，子线程才能看到当前页面<br /> cur_buf->exch_usr_cnt = noofchild;<br /> MEfill(sizeof(cur_buf->exch_visit), '/0', &cur_buf->exch_visit);</p> <p> /* Is it last buffer */<br /> if (cur_buf->exch_bufno == exch_buf_hdr->ehdr_noofbufs)<br /> {<br /> /* set the next buffer to 1 */<br /> next_buf = exch_buf_hdr->ehdr_exch_buffers;<br /> }<br /> else<br /> {<br /> next_buf = cur_buf;<br /> next_buf++;<br /> }<br /> CSv_semaphore(&cur_buf->exch_buf_mutex);</p> <p> CSp_semaphore(TRUE, &next_buf->exch_buf_mutex);<br /> //下面的保证，空闲页面必须是每个子线程读过的<br /> while ( exch_buf_hdr->ehdr_error_flag ||<br /> next_buf->exch_usr_cnt > 0 )<br /> {<br /> /*<br /> ** check the error code here for children & if they wake<br /> ** you up for error<br /> */<br /> if (exch_buf_hdr->ehdr_error_flag)<br /> {<br /> #ifdef xDEBUG<br /> TRdisplay("Parent saw the error and going to exit routines for children to finish/n");<br /> #endif<br /> /* we have now seen child raised error ... hence exit */<br /> CSv_semaphore(&next_buf->exch_buf_mutex);<br /> return(E_DB_ERROR);<br /> }</p> <p> /* Tell children we're waiting on this buffer */<br /> exch_buf_hdr->ehdr_parent_status = next_buf->exch_bufno;<br /> CSv_semaphore(&next_buf->exch_buf_mutex);<br /> CSsuspend(0,0,0);<br /> /* CSresume-er will clear ehdr_parent_status */<br /> CSp_semaphore(TRUE, &next_buf->exch_buf_mutex);</p> <p> /* Cancel any extra resumes from those pesky kids */<br /> CScancelled((PTR)0);<br /> }<br /> //清空页面中存入的元组数<br /> next_buf->exch_noofrec = 0;<br /> MEfill(sizeof(next_buf->exch_visit), '/0', &next_buf->exch_visit);<br /> *cur_buf_ptr = next_buf; </p> <p> return(E_DB_OK);<br /> }

实现了一个线程写，多个线程读，每个读线程读到全是一样的，