多线程 操作多文件 性能方案讨论与请教

Question

前不久刚完成了一个IOCP的服务器现在碰到一个问题就是该IOCP会操作很多的文件  (文件个数可能会是上万个)  其中可能会有100来个 更新会很频繁
然后做2种操作 一个是读   一个是创建文件  (已存在的文件可能会被重新创建)现在问题就是  如果只用一个临界区做锁 那么 当多个线程都为读的时候 性能又会很差  因为不管是读多个文件还是读同一个文件 其实都是可以同时去读的 并不需要去锁但更不可能为每个文件都去创建锁  如果取消锁 拼运气(读的频率会很高 写的频率相对会低很多) 那又感觉好像太不厚道了...
所以向大家请教与讨论  寻求更好的解决方案

Answer 1

这个问题最大的难度不在于加锁，而在于允许多个线程同时读同一个文件（即读共享），此时一个试图写该文件的线程必须等待所有读该文件的线程全部退出之后才能开始写，而Windows却并没有提供这种等待机制，除非自己做线程调度。

Answer 2

你的看法是最明白的  目前我也是没找到很完美的办法  因为有几个重点问题
1、是大量的文件 如果上万的文件 因此我不想建立上万个临界区
2、多线程也可能操作的时候（不管读还是写） 因为文件多 可能读写的重复率相对较低 只用一个临界区 肯定效率低 我也想的到允许并发读 写的时候锁
正如2楼所讲 Windows没有提供现成的方法  你来展示一下完整的思路和解决方案吧  可以用伪代码描述一下  

Answer 3

你的这个思路我也考虑过，乍一想没什么问题，可如果你真要去这么实现，就会发现问题不是一般的多。比如一个读线程，它必须先检测是否存在写锁，如果有的话等待，没有的话要给读计数加一，然后又要检测有没有读锁，没有的话要新建一个（必须给写线程建立一个可以用来Wait的东西）。这些操作必须放入一个临界区，否则等你刚做完判断，还没来得及执行操作，别的线程就把你的判断依据给改了。这个临界区是所有线程共用的一个单一临界区（进临界区之前还没说明你要操作哪一个文件呢，所以任一线程对任一文件操作的时候，都得到这一个临界区排队），因此绝不能有线程在这个临界区里Wait。既然如此，那么你检测锁的操作，都只能是检测，比如读线程最初检测是不是有写锁存在的时候，如果检测到有的话，那么必须先退出临界区，然后再Wait写锁。可是等你Wait完了呢？还得再进临界区来设置读计数等等（还得记得释放写锁，否则后面的读线程就进不来了）！如果看到这里你还没晕的话，恭喜你，你可以再去考虑一下写线程的细节问题了不出意外的话，当你想清楚写线程的问题后，你会发现前面关于读线程的思路尚需要进一步调整一下，嗯，确切地说，是进一步复杂化一下
总之，除非你放弃读共享（即：不允许多个线程同时读同一个文件），否则的话，最合理的方式是建立一个专门的调度线程，用来控制所有其它线程在文件操作上的同步和互斥。单靠线程本身自发控制，就算你现在能彻底把这团乱麻想清楚，回头过几天你自己就又看不懂了。其实，文件的读写共享问题，操作系统本身就可以处理得很好，只不过当共享冲突的时候，CreateFile并不等待，而是直接返回错误。而你现在的问题无非就是要解决这个等待吗？那你弄个循环，只要发生共享冲突的时候Sleep(50)就是了，虽然粗糙点，不过这实在是最简单的解决方案了。

Answer 4

摘自 <MultiThreading Application In Win32>网上有侯捷翻译的中文版<Win32多线程程序设计>ReadWrite.h/*
 * ReadWrit.h
 *
 * Sample code for Multithreading Applications in Win32
 * This is from Chapter 7, Listing 7-1
 *
 * Demonstrates an implementation of the
 * Readers/Writers algorithm. This version
 * gives preference to readers.
 *//////////////////////////////////////////////////////////
// Structure definition
//typedef struct _RWLock
{
    // Handle to a mutex that allows
    // a single reader at a time access
    // to the reader counter.
    HANDLE hMutex;    // Handle to a semaphore that keeps
    // the data locked for either the
    // readers or the writers.
    HANDLE hDataLock;    // The count of the number of readers.
    // Can legally be zero or one while
    // a writer has the data locked.
int nReaderCount;
} RWLock;//
// Reader/Writer prototypes
//BOOL InitRWLock(RWLock *pLock);
BOOL DestroyRWLock(RWLock *pLock);
BOOL AcquireReadLock(RWLock *pLock);
int ReleaseReadLock(RWLock *pLock);
BOOL AcquireWriteLock(RWLock *pLock);
int ReleaseWriteLock(RWLock *pLock);
BOOL ReadOK(RWLock *pLock);
BOOL WriteOK(RWLock *pLock);BOOL FatalError(char *s);

Answer 5

ReadWrite.c/*
 * ReadWrit.c
 *
 * Sample code for "Multithreading Applications in Win32"
 * This is from Chapter 7, various listings.
 *
 * Demonstrates an implementation of the
 * Readers/Writers algorithm. This version
 * gives preference to readers.
 */#define WIN32_LEAN_AND_MEAN
#include <stdio.h>
#include <stdlib.h>
#include <windows.h>
#include "ReadWrit.h"// If we wait more than 2 seconds, then something is probably wrong!
#define MAXIMUM_TIMEOUT 2000// Here's the pseudocode for what is going on:
//
// Lock for Reader:
//  Lock the mutex
//  Bump the count of readers
//  If this is the first reader, lock the data
//  Release the mutex
//
// Unlock for Reader:
//  Lock the mutex
//  Decrement the count of readers
//  If this is the last reader, unlock the data
//  Release the mutex
//
// Lock for Writer:
//  Lock the data
//
// Unlock for Reader:
//  Unlock the data///////////////////////////////////////////////////////BOOL MyWaitForSingleObject(HANDLE hObject)
{
    DWORD result;    result = WaitForSingleObject(hObject, MAXIMUM_TIMEOUT);
    // Comment this out if you want this to be non-fatal
    if (result != WAIT_OBJECT_0)
        FatalError("MyWaitForSingleObject - Wait failed, you probably forgot to call release!");
    return (result == WAIT_OBJECT_0);
}BOOL InitRWLock(RWLock *pLock)
{
    pLock->nReaderCount = 0;
    pLock->hDataLock = CreateSemaphore(NULL, 1, 1, NULL);
    if (pLock->hDataLock == NULL)
        return FALSE;
    pLock->hMutex = CreateMutex(NULL, FALSE, NULL);
    if (pLock->hMutex == NULL)
    {
        CloseHandle(pLock->hDataLock);
        return FALSE;
    }
    return TRUE;
}BOOL DestroyRWLock(RWLock *pLock)
{
    DWORD result = WaitForSingleObject(pLock->hDataLock, 0);
    if (result == WAIT_TIMEOUT)
        return FatalError("DestroyRWLock - Can't destroy object, it's locked!");    CloseHandle(pLock->hMutex);
    CloseHandle(pLock->hDataLock);
    return TRUE;
}BOOL AcquireReadLock(RWLock *pLock)
{
    BOOL result = TRUE;    if (!MyWaitForSingleObject(pLock->hMutex))
        return FALSE;
 
    if(++pLock->nReaderCount == 1)
        result = MyWaitForSingleObject(pLock->hDataLock);    ReleaseMutex(pLock->hMutex);
    return result;
}BOOL ReleaseReadLock(RWLock *pLock)
{
    int result;
    LONG lPrevCount;    if (!MyWaitForSingleObject(pLock->hMutex))
        return FALSE;    if (--pLock->nReaderCount == 0)
        result = ReleaseSemaphore(pLock->hDataLock, 1, &lPrevCount);    ReleaseMutex(pLock->hMutex);
    return result;
}BOOL AcquireWriteLock(RWLock *pLock)
{
    return MyWaitForSingleObject(pLock->hDataLock);
}BOOL ReleaseWriteLock(RWLock *pLock)
{
    int result;
    LONG lPrevCount;    result = ReleaseSemaphore(pLock->hDataLock, 1, &lPrevCount);
    if (lPrevCount != 0)
        FatalError("ReleaseWriteLock - Semaphore was not locked!");
    return result;
}BOOL ReadOK(RWLock *pLock)
{
    // This check is not perfect, because we
    // do not know for sure if we are one of
    // the readers.
    return (pLock->nReaderCount > 0);
}BOOL WriteOK(RWLock *pLock)
{
    DWORD result;    // The first reader may be waiting in the mutex,
    // but any more than that is an error.
    if (pLock->nReaderCount > 1)
        return FALSE;    // This check is not perfect, because we
    // do not know for sure if this thread was
    // the one that had the semaphore locked.
    result = WaitForSingleObject(pLock->hDataLock, 0);
    if (result == WAIT_TIMEOUT)
        return TRUE;    // a count is kept, which was incremented in Wait.
    result = ReleaseSemaphore(pLock->hDataLock, 1, NULL);
    if (result == FALSE)
        FatalError("WriteOK - ReleaseSemaphore failed");
    return FALSE;
}////////////////////////////////////////////////////////*
 * Error handler
 */
BOOL FatalError(char *s)
{
    fprintf(stdout, "%s\n", s);
    // Comment out exit() to prevent termination
    exit(EXIT_FAILURE);
    return FALSE;
}
测试程序/*
 * List.c
 *
 * Sample code for "Multithreading Applications in Win32"
 * This is from Chapter 7, Listing 7-1
 *
 * Demonstrates an implementation of the
 * Readers/Writers algorithm. This version
 * gives preference to readers.
 */#define WIN32_LEAN_AND_MEAN
#include <stdio.h>
#include <stdlib.h>
#include <windows.h>
#include "ReadWrit.h"/////////////////////////////////////////////////////////
// Structure definition
//typedef struct _Node
{
    struct _Node *pNext;
    char szBuffer[80];
} Node;typedef struct _List
{
RWLock lock;
    Node *pHead;
} List;//
// Linked list prototypes
//BOOL InitRWLock(RWLock *pLock);
BOOL DeleteList(List *pList);
BOOL AddHead(List *pList, Node *node);
BOOL DeleteHead(List *pList);
BOOL Insert(List *pList, Node *afterNode, Node *newNode);
Node *Next(List *pList, Node *node);//
// Test functions prototypes
//DWORD WINAPI LoadThreadFunc(LPVOID n);
DWORD WINAPI SearchThreadFunc(LPVOID n);
DWORD WINAPI DeleteThreadFunc(LPVOID n);//
// Global variables
//// This is the list we use for testing
List *gpList;///////////////////////////////////////////////////////List *CreateList()
{
    List *pList = GlobalAlloc(GPTR, sizeof(List));
if (InitRWLock(&pList->lock) == FALSE)
{
GlobalFree(pList);
pList = NULL;
}
    return pList;
}BOOL DeleteList(List *pList)
{
AcquireWriteLock(&pList->lock);
while (DeleteHead(pList))
;
ReleaseWriteLock(&pList->lock); DestroyRWLock(&gpList->lock);    GlobalFree(pList); return TRUE;
}BOOL AddHead(List *pList, Node *pNode)
{
if (!WriteOK(&pList->lock))
return FatalError("AddHead - not allowed to write!");    pNode->pNext = pList->pHead;
    pList->pHead = pNode;
}BOOL DeleteHead(List *pList)
{
Node *pNode; if (!WriteOK(&pList->lock))
return FatalError("AddHead - not allowed to write!"); if (pList->pHead == NULL)
return FALSE; pNode = pList->pHead->pNext;
GlobalFree(pList->pHead);
    pList->pHead = pNode;
return TRUE;
}BOOL Insert(List *pList, Node *afterNode, Node *newNode)
{
if (!WriteOK(&pList->lock))
return FatalError("Insert - not allowed to write!");    if (afterNode == NULL)
    {
        AddHead(pList, newNode);
    }
    else
    {
        newNode->pNext = afterNode->pNext;
        afterNode->pNext = newNode;
    }
}Node *Next(List *pList, Node *pNode)
{
if (!ReadOK(&pList->lock))
{
FatalError("Next - Not allowed to read!");
return NULL;
} if (pNode == NULL)
return pList->pHead;
else
    return pNode->pNext;
}///////////////////////////////////////////////////////DWORD WINAPI ThreadFunc(LPVOID);int main()
{
    HANDLE  hThrds[4];
    int     slot = 0;
int rc;
int nThreadCount = 0;
    DWORD dwThreadId; gpList = CreateList();
if (!gpList)
FatalError("main - List creation failed!");    hThrds[nThreadCount++] = CreateThread(NULL,
        0, LoadThreadFunc, 0, 0, &dwThreadId );    hThrds[nThreadCount++] = CreateThread(NULL,
        0, SearchThreadFunc, (LPVOID)"pNode", 0, &dwThreadId );    hThrds[nThreadCount++] = CreateThread(NULL,
        0, SearchThreadFunc, (LPVOID)"pList", 0, &dwThreadId );    hThrds[nThreadCount++] = CreateThread(NULL,
        0, DeleteThreadFunc, 0, 0, &dwThreadId );    /* Now wait for all threads to terminate */
    rc = WaitForMultipleObjects(
        nThreadCount,
        hThrds,
        TRUE,
        INFINITE );    for (slot=0; slot<nThreadCount; slot++)
        CloseHandle(hThrds[slot]);
    printf("\nProgram finished.\n"); DeleteList(gpList);    return EXIT_SUCCESS;
}/*
 * Slowly load the contents of "List.c" into the
 * linked list.
 */
DWORD WINAPI LoadThreadFunc(LPVOID n)
{
int nBatchCount;
Node *pNode; FILE* fp = fopen("List.c", "r");
if (!fp)
{
fprintf(stderr, "ReadWrit.c not found\n");
exit(EXIT_FAILURE);
} pNode = GlobalAlloc(GPTR, sizeof(Node));
nBatchCount = (rand() % 10) + 2;
AcquireWriteLock(&gpList->lock); while (fgets(pNode->szBuffer, sizeof(Node), fp))
{
AddHead(gpList, pNode); // Try not to hog the lock
if (--nBatchCount == 0)
{
ReleaseWriteLock(&gpList->lock);
Sleep(rand() % 5);
nBatchCount = (rand() % 10) + 2;
AcquireWriteLock(&gpList->lock);
}
pNode = GlobalAlloc(GPTR, sizeof(Node));
} ReleaseWriteLock(&gpList->lock);
return 0;
}
/*
 * Every so often, walked the linked list
 * and figure out how many lines one string
 * appears (given as the startup param)
 */
DWORD WINAPI SearchThreadFunc(LPVOID n)
{
int i;
char *szSearch = (char *)n; for (i=0; i<20; i++)
{
int nFoundCount = 0;
Node   *next = NULL; AcquireReadLock(&gpList->lock);
next = Next(gpList, next);
while (next)
{
if (strstr(next->szBuffer, szSearch))
nFoundCount++;
next = Next(gpList, next);
} ReleaseReadLock(&gpList->lock); printf("Found %d lines with '%s'\n", nFoundCount, szSearch);
Sleep((rand() % 30));
}
    return 0;
}
/*
 * Every so often, delete some entries in the list.
 */
DWORD WINAPI DeleteThreadFunc(LPVOID n)
{
int i; for (i=0; i<100; i++)
{
Sleep(1);
AcquireWriteLock(&gpList->lock);
DeleteHead(gpList);
DeleteHead(gpList);
DeleteHead(gpList);
ReleaseWriteLock(&gpList->lock);
}    return 0;
}