标签: semaphore

我正在编写一个模仿电影院的多线程应用程序.涉及的每个人都是自己的线程,并发必须完全由信号量完成.我唯一的问题是如何基本上链接线程,以便他们可以通信(例如通过管道).

例如:

客户[1]是一个线程,获取一个信号量,让它走向票房.现在,客户[1]必须告诉票房代理他们想要看电影"X".然后BoxOfficeAgent [1]也是一个线程,必须检查以确保电影未满,并且要么卖票,要么告诉客户[1]选择另一部电影.

如何在保持与信号量的并发性的同时来回传递数据？

另外,我可以在java.util.concurrent中使用的唯一类是Semaphore类.

这是我的问题:我有一个脚本(让我们称之为comet.php)由AJAX客户端脚本提供,并等待更改发生如下:

while(no_changes){
    usleep(100000);
    //check for changes
}

我不太喜欢这个,它不是很可扩展,它是(imho)"糟糕的做法"我想用信号量(？)或无论如何并发编程技术来改善这种行为.你能告诉我一些如何处理这个问题的技巧吗？(我知道,这不是一个简短的答案,但一个起点就足够了.)

编辑:LibEvent怎么样？

如果相同代码的多个实例在不同的服务器上运行,我想使用数据库来确保如果某个服务器已经在另一台服务器上运行,则它不会在一台服务器上启动.

我可能会提出一些可行的SQL命令,它们使用Oracle事务处理,锁存器或其他任何东西,但我宁愿找到一些经过尝试和真实的东西.

几年前,作为SQL wiz的开发人员有一个SQL事务,它获取了信号量,如果得到它则返回true,如果没有得到它则返回false.然后在我处理结束时,我需要运行另一个SQL事务来释放信号量.这很酷,但我不知道数据库支持的信号量是否有可能超时.超时是一个巨大的好处!

编辑:

以下是一些可行的SQL命令,但除了通过cron job hack之外没有超时:

---------------------------------------------------------------------
--Setup
---------------------------------------------------------------------
CREATE TABLE "JOB_LOCKER" ( "JOB_NAME" VARCHAR2(128 BYTE), "LOCKED" VARCHAR2(1 BYTE), "UPDATE_TIME" TIMESTAMP (6) );
CREATE UNIQUE INDEX "JOB_LOCKER_PK" ON "JOB_LOCKER" ("JOB_NAME") ;
ALTER TABLE "JOB_LOCKER" ADD CONSTRAINT "JOB_LOCKER_PK" PRIMARY KEY ("JOB_NAME");
ALTER TABLE "JOB_LOCKER" MODIFY ("JOB_NAME" NOT NULL ENABLE);
ALTER TABLE "JOB_LOCKER" MODIFY ("LOCKED" NOT NULL ENABLE);

insert into job_locker (job_name, locked) values ('myjob','N');
commit;

---------------------------------------------------------------------
--Execute at the beginning of the job
--AUTOCOMMIT MUST BE OFF!
---------------------------------------------------------------------
select …

当 C# 程序持有命名信号量时，如果应用程序提前终止（例如通过按 Ctrl+C 或关闭控制台窗口），它似乎不会被释放。至少在进程的所有实例都终止之前不会。

在这种情况下，使用命名互斥量会引发 AbandonedMutexException，但不会引发信号量。当另一个程序实例提前终止时，如何防止一个程序实例停顿？

class Program
{
    // Same with count > 1
    private static Semaphore mySemaphore = new Semaphore(1, 1, "SemaphoreTest");

    static void Main(string[] args)
    {
        try
        {
            // Blocks forever if the first process was terminated
            // before it had the chance to call Release
            Console.WriteLine("Getting semaphore");
            mySemaphore.WaitOne();  
            Console.WriteLine("Acquired...");
        }
        catch (AbandonedMutexException)
        {
            // Never called!
            Console.WriteLine("Acquired due to AbandonedMutexException...");
        }
        catch (System.Exception ex)
        {
            Console.WriteLine(ex);
        }

        Thread.Sleep(20 * 1000);
        mySemaphore.Release();
        Console.WriteLine("Done");
    }
}

我正在尝试调解一个小的Windows服务,使其在启动期间等待来自另一个进程的信号.当然,我知道这种方法可能(甚至会)有时会导致服务启动超时.事实并非如此.

问题是名为System.Thread.Sempaphore我用于调解目的.使用以下构造在其他地方创建和获取信号量.没有更改GC有它,因为我明确地在给定的行下方执行测试以进行测试.

Boolean newOne;
System.Threading.Semaphore rootSemaphore = 
    new System.Threading.Semaphore(1, 1, "DummyServiceSemaphore", out newOne);

上面的代码显然效果很好.以下代码在调试模式或控制台应用程序下执行时效果很好:

Boolean createdNew;
System.Threading.Semaphore semaphore = 
    new System.Threading.Semaphore(1, 1, "DummyServiceSemaphore", out createdNew);
if (createdNew) 
    throw new Exception("That's not what we wanted");

作为Windows服务的一部分执行时,完全相同的代码失败:

static class Program
{
    static void Main(string[] args)
    {
        Boolean createdNew;
        System.Threading.Semaphore semaphore = 
            new System.Threading.Semaphore(1, 1, "DummyServiceSemaphore", out createdNew);
        if (createdNew) 
            throw new Exception("That's not what we wanted");

        ServiceBase[] ServicesToRun;
        ServicesToRun = new ServiceBase[] { new Dummy() };
        ServiceBase.Run(ServicesToRun);
    }
}

所以,请寻求帮助.

PS:我一直在尝试使用Mutex,但是还有另外一个问题 - …

我必须使用Java信号量解决这个问题,但我不知道如何,我找不到任何相关的Java资料.这是怎么回事:

有各种各样的线索:男人和女人.两者都想使用数量为BATHROOM_SIZE的相同资源.5条规则:

1. 在信令需要使用资源之后,每个线程都应该等到他能够使用它.
2. 当多个BATHOOM_SIZE线程同时使用资源时,防止出现这种情况.
3. 防止女人和男人同时使用浴室.
4. 线程应该同时使用资源.如果有一种类型的线程很多,那么BATHROOM_SIZE线程应该使用资源.
5. 防止饥饿.

结果

效劳于:

1女,1人,5女,5人

失败:

5women1men,5men1women,2men2women,5men5women.

自从星期一以来我一直在努力使它成功,现在我的想法已经用完了.

码

所以我的任务是编写实现BathroomInterface的Bathroom.java类:

public interface BathroomInterface {

    public static final int BATHROOM_SIZE = 3; //3 is just example
    void manEnter();
    void manExit();
    void womanEnter();
    void womanExit();
}

在系统中有许多男人和女人的线程像这样工作:

for(int i = 0; i < n; i++) {
  bathroom.manEnter();
  //uses bathroom random amount of time
  bathroom.manExit();
}

for(int i = 0; i < m; i++) {
  bathroom.womanEnter();
  //uses bathroom random amount of time
  bathroom.womanExit();
}

我也有 …

我有两个进程A和B. A和B有时需要进行通信(双向)以传递信号,消息等.
我已经对Linux中可用的IPC进行了一些基础研究,如信号量,消息队列,dbus等.
现在我是在决定使用哪一个时很困惑,谁能告诉我哪个IPC更适合我的应用？

提前致谢

编辑:阐述应用程序.(这是一个嵌入式应用程序)
过程A将监控温度,速度计算等.过程B将驱动电机,读取传感器值(数字)等.有时我需要将信号发送到过程B告知达到最高温度,所以停止驱动电机.有时需要将从流程A中的传感器读取的数据发送到流程B.像这样,需要在流程中传递数字数据.我在ARM架构中这样做.

对于当前的OpenCL GPGPU项目,我需要根据具有64个可能值的某个键对数组中的元素进行排序.我需要最后一个数组让所有具有相同键的元素都是连续的.将关联数组new_index[old_index]作为此任务的输出就足够了.

我把任务分成两部分.首先,我计算每个可能的密钥(桶)使用此密钥(进入该桶)的元素数量.我扫描这个数组(生成一个前缀和),它指示每个桶的元素的新索引范围,例如每个桶的"开始"索引.

然后,第二步必须为每个元素分配一个新索引.如果我要在CPU上实现它,算法将是这样的:

for all elements e:
    new_index[e] = bucket_start[bucket(e)]++

当然,这在GPU上不起作用.每个项目都需要以bucket_start读写模式访问数组,这实际上是所有工作项之间的同步,这是我们能做的最糟糕的事情.

一个想法是在工作组中进行一些计算.但我不确定应该如何完成,因为我没有GPGPU计算经验.

在全局内存中,我们使用前缀sum初始化了bucket start数组,如上所述.使用原子int"访问"此数组的"互斥".(我是新来的,所以也许在这里混合一些词.)

每个工作组都隐式分配了输入元素数组的一部分.它使用包含新索引的本地存储桶数组,相对于我们尚不知道的(全局)存储桶启动.在其中一个"本地缓冲区"已满后,工作组必须将本地缓冲区写入全局数组.为此,它锁定对全局存储区启动数组的访问,将这些值递增当前本地存储区大小,解锁,然后将结果写入全局new_index数组(通过添加相应的偏移量).重复此过程,直到处理完所有已分配的元素.

出现两个问题:

1. 这是一个好方法吗？我知道从全局内存读取和写入很可能是这里的瓶颈,特别是因为我正在尝试获取全局内存的(至少只有一小部分)同步访问.但也许有更好的方法来做到这一点,也许使用内核分解.请注意,我尝试避免在内核期间将数据从GPU读回到CPU(以避免OpenCL命令队列刷新,这也是我所采用的那样糟糕).

2. 在上面的算法设计中,我该如何实现锁定机制？以下代码会起作用吗？特别是,当硬件在SIMD组中执行工作项"真正并行"时,我会遇到问题,比如Nvidia"warps".在我当前的代码中,工作组的所有项目都会尝试以SIMD方式获取锁定.我应该仅限于第一个工作项吗？并使用障碍使他们在本地保持同步？

   #pragma OPENCL EXTENSION cl_khr_global_int32_base_atomics : enable
   
   __kernel void putInBuckets(__global uint *mutex,
                              __global uint *bucket_start,
                              __global uint *new_index)
   {
       __local bucket_size[NUM_BUCKETS];
       __local bucket[NUM_BUCKETS][LOCAL_MAX_BUCKET_SIZE]; // local "new_index"
   
       while (...)
       {
           // process a couple of elements locally until a local bucket is full
           ...
   
           // "lock"
           while(atomic_xchg(mutex, 1)) {
           }
   
           // "critical section"
           __local …

   Run Code Online (Sandbox Code Playgroud)

任何解释每个的实时场景都将受到赞赏.除了pthreads之外还有其他方法来处理同步.互斥体与递归互斥体(任何实时场景)有何不同？

由于使用了太多的Websphere Message Broker实例或者某些东西,我们的Linux机箱上的信号量用完了.

一位同事和我想知道为什么这甚至是有限的 - 它只是一点记忆,对吧？

我彻底用Google搜索,一无所获.

有谁知道这是为什么？

干杯