Notes/fundamentals
1
0
Fork 0
mirror of https://github.com/robindhole/fundamentals.git synced 2025-03-15 16:50:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Adds semaphores.

Browse Source
This commit is contained in:
Tanmay 2022-08-24 13:57:51 +01:00
parent c7437d90bc
commit 7dbf8cb992
3 changed files with 35 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
os/code/os/src/main/java/com/scaler/producerconsumer/Consumer.java
View File

@ -1,6 +1,7 @@
package com.scaler.producerconsumer;
import java.util.Queue;
import java.util.concurrent.Semaphore;
import lombok.AllArgsConstructor;
import lombok.Getter;
@ -11,16 +12,25 @@ public class Consumer implements Runnable {
private Queue<UnitOfWork> store;
private String name;
private Semaphore forProducer;
private Semaphore forConsumer;
@Override
public void run() {
while (true) {
synchronized (store) {
if (store.size() > 0) {
store.remove();
System.out.println("Consumed: " + name + " Left units :" + store.size());
}
try {
forConsumer.acquire();
} catch (InterruptedException e) {
throw new RuntimeException("Error acquiring semaphore " + e);
}
if (store.size() > 0) {
store.remove();
System.out.println("Consumed: " + name + " Left units :" + store.size());
}
forProducer.release();
}
}

19
os/code/os/src/main/java/com/scaler/producerconsumer/Producer.java
View File

@ -1,6 +1,7 @@
package com.scaler.producerconsumer;
import java.util.Queue;
import java.util.concurrent.Semaphore;
import lombok.AllArgsConstructor;
import lombok.Getter;
@ -12,16 +13,24 @@ public class Producer implements Runnable {
private Queue<UnitOfWork> store;
private int maxSize;
private String name;
private Semaphore forProducer;
private Semaphore forConsumer;
@Override
public void run() {
while (true) {
synchronized (store) {
if (store.size() < maxSize) {
store.add(new UnitOfWork());
System.out.println("Produced: " + name + " Left units :" + store.size());
}
try {
forProducer.acquire();
} catch (InterruptedException e) {
throw new RuntimeException("Error acquiring semaphore " + e);
}
if (store.size() < maxSize) {
store.add(new UnitOfWork());
System.out.println("Produced: " + name + " Left units :" + store.size());
}
forConsumer.release();
}
}

8
os/code/os/src/main/java/com/scaler/producerconsumer/Runner.java
View File

@ -3,6 +3,7 @@ package com.scaler.producerconsumer;
import java.util.ArrayDeque;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.Semaphore;
import java.util.stream.Collectors;
public class Runner {
@ -14,14 +15,17 @@ public class Runner {
Queue<UnitOfWork> store = new ArrayDeque<>();
int maxSize = 20;
Semaphore forProducer = new Semaphore(maxSize);
Semaphore forConsumer = new Semaphore(0);
Set<Producer> producers = producerNames
.stream()
.map(name -> new Producer(store, maxSize, name))
.map(name -> new Producer(store, maxSize, name, forProducer, forConsumer))
.collect(Collectors.toSet());
Set<Consumer> consumers = consumerNames
.stream()
.map(name -> new Consumer(store, name))
.map(name -> new Consumer(store, name, forProducer, forConsumer))
.collect(Collectors.toSet());
producers.forEach(producer -> new Thread(producer).start());