12:35 a.m.
I am using Drools Fusion 6.0.1.Final version, it looks like some (rather
several) actions to the events are dropped. I am sending about 100K events
and expecting an action to all the events, but only 5000 actions are
performed, which means about 95K actopms are dropped.
The same issue was posted in the thread
http://drools.46999.n3.nabble.com/rules-users-Drools-Fusion-Dropping-Even...,
but it is marked as resolved.
The defect https://issues.jboss.org/browse/DROOLS-131 is also marked as
fixed.
But this issue seems to be appearing again.
Here is my code.
kmodule.xml
<?xml version="1.0" encoding="UTF-8"?>
<kmodule xmlns="http://jboss.org/kie/6.0.0/kmodule">
<kbase name="eventDrop" packages="ed"
eventProcessingMode="stream">
<ksession name="edSession"/>
</kbase>
</kmodule>
ed.drl
package drools
import foo.TemperatureEvent
declare TemperatureEvent
@role(event)
end
rule "RULE S1"
when
$te:TemperatureEvent(measure <= 50) from entry-point entryone
then
System.err.println("The temperature at location - "+$te.getLocation() +
" is " + $te.getMeasure());
end
rule "RULE S2"
when
$te:TemperatureEvent(measure > 50) from entry-point entryone
then
System.err.println("The temperature at location - "+$te.getLocation() +
" is " + $te.getMeasure());
end
TemperatureEvent.java
package foo;
public class TemperatureEvent {
int measure;
String location;
long timeStamp;
public TemperatureEvent(int measure, String location, long timeStamp) {
this.measure = measure;
this.location = location;
this.timeStamp = timeStamp;
}
public int getMeasure() {
return measure;
}
public void setMeasure(int measure) {
this.measure = measure;
}
public String getLocation() {
return location;
}
public void setLocation(String location) {
this.location = location;
}
public long getTimeStamp() {
return timeStamp;
}
public void setTimeStamp(long timeStamp) {
this.timeStamp = timeStamp;
}
}
App.java
package foo;
import java.io.FileOutputStream;
import java.io.PrintStream;
import java.util.Random;
import org.kie.api.KieServices;
import org.kie.api.runtime.KieContainer;
import org.kie.api.runtime.KieSession;
import org.kie.api.runtime.rule.EntryPoint;
public class App {
public static void main(String[] args) throws Exception {
System.setErr(new PrintStream(new FileOutputStream("report.txt")));
// Start Drools.
KieServices ks = KieServices.Factory.get();
KieContainer kContainer = ks.getKieClasspathContainer();
final KieSession kSession = kContainer.newKieSession("edSession");
EntryPoint entryPoint1 = kSession.getEntryPoint("entryone");
new Thread() {
@Override
public void run() {
kSession.fireUntilHalt();
}
}.start();
Random measure = new Random(System.currentTimeMillis());
Random location = new Random(System.currentTimeMillis() - 1000000L);
String locations[] = { "Bangalore", "Chennai",
"Delhi", "Mumbai",
"Kolkata" };
for (int i = 0; i < 10; ++i) {
for (int j = 0; j < 1000; ++j) {
TemperatureEvent te = new TemperatureEvent(
measure.nextInt(100),
locations[location.nextInt(5)],
System.currentTimeMillis());
entryPoint1.insert(te);
// Thread.sleep(1);
}
}
Thread.sleep(100);
kSession.halt();
kSession.dispose();
}
}
--
View this message in context:
http://drools.46999.n3.nabble.com/Drools-Fusion-Dropping-Actions-to-Event...
Sent from the Drools: User forum mailing list archive at Nabble.com.
1:43 a.m.
This seems to be a race condition, the KIE session not being thread-safe:
if events are inserted in a tight loop, about 50% simply evaporate.
Using Thread.sleep(1) after each insert avoids all losses, but that's
hardly a
feasible workaround, since this doesn't guarantee that the thread running
the session has become active and "digested" the fact.
-W
On 22/04/2014, chandu_divi <chandu_divi(a)hotmail.com> wrote:
I am using Drools Fusion 6.0.1.Final version, it looks like some
(rather
several) actions to the events are dropped. I am sending about 100K events
and expecting an action to all the events, but only 5000 actions are
performed, which means about 95K actopms are dropped.
The same issue was posted in the thread
http://drools.46999.n3.nabble.com/rules-users-Drools-Fusion-Dropping-Even...,
but it is marked as resolved.
The defect https://issues.jboss.org/browse/DROOLS-131 is also marked as
fixed.
But this issue seems to be appearing again.
Here is my code.
kmodule.xml
<?xml version="1.0" encoding="UTF-8"?>
<kmodule xmlns="http://jboss.org/kie/6.0.0/kmodule">
<kbase name="eventDrop" packages="ed"
eventProcessingMode="stream">
<ksession name="edSession"/>
</kbase>
</kmodule>
ed.drl
package drools
import foo.TemperatureEvent
declare TemperatureEvent
@role(event)
end
rule "RULE S1"
when
$te:TemperatureEvent(measure <= 50) from entry-point entryone
then
System.err.println("The temperature at location - "+$te.getLocation() +
" is " + $te.getMeasure());
end
rule "RULE S2"
when
$te:TemperatureEvent(measure > 50) from entry-point entryone
then
System.err.println("The temperature at location - "+$te.getLocation() +
" is " + $te.getMeasure());
end
TemperatureEvent.java
package foo;
public class TemperatureEvent {
int measure;
String location;
long timeStamp;
public TemperatureEvent(int measure, String location, long timeStamp) {
this.measure = measure;
this.location = location;
this.timeStamp = timeStamp;
}
public int getMeasure() {
return measure;
}
public void setMeasure(int measure) {
this.measure = measure;
}
public String getLocation() {
return location;
}
public void setLocation(String location) {
this.location = location;
}
public long getTimeStamp() {
return timeStamp;
}
public void setTimeStamp(long timeStamp) {
this.timeStamp = timeStamp;
}
}
App.java
package foo;
import java.io.FileOutputStream;
import java.io.PrintStream;
import java.util.Random;
import org.kie.api.KieServices;
import org.kie.api.runtime.KieContainer;
import org.kie.api.runtime.KieSession;
import org.kie.api.runtime.rule.EntryPoint;
public class App {
public static void main(String[] args) throws Exception {
System.setErr(new PrintStream(new FileOutputStream("report.txt")));
// Start Drools.
KieServices ks = KieServices.Factory.get();
KieContainer kContainer = ks.getKieClasspathContainer();
final KieSession kSession = kContainer.newKieSession("edSession");
EntryPoint entryPoint1 = kSession.getEntryPoint("entryone");
new Thread() {
@Override
public void run() {
kSession.fireUntilHalt();
}
}.start();
Random measure = new Random(System.currentTimeMillis());
Random location = new Random(System.currentTimeMillis() -
1000000L);
String locations[] = { "Bangalore", "Chennai",
"Delhi", "Mumbai",
"Kolkata" };
for (int i = 0; i < 10; ++i) {
for (int j = 0; j < 1000; ++j) {
TemperatureEvent te = new TemperatureEvent(
measure.nextInt(100),
locations[location.nextInt(5)],
System.currentTimeMillis());
entryPoint1.insert(te);
// Thread.sleep(1);
}
}
Thread.sleep(100);
kSession.halt();
kSession.dispose();
}
}
--
View this message in context:
http://drools.46999.n3.nabble.com/Drools-Fusion-Dropping-Actions-to-Event...
Sent from the Drools: User forum mailing list archive at Nabble.com.
_______________________________________________
rules-users mailing list
rules-users(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/rules-users
3:17 a.m.
Thanks a lot for the information.
Adding 'Thread.sleep(1);' after every insert reduces the problem intensity,
but sometimes even this solution also does not perform all the expected
actions (May be some events are dropped due to race condition).
Is there a way I can wait for KIE session to digest the event before I can
insert another event?
Thanks,
Chandra.
--
View this message in context:
http://drools.46999.n3.nabble.com/Drools-Fusion-Dropping-Actions-to-Event...
Sent from the Drools: User forum mailing list archive at Nabble.com.
4:10 a.m.
On 22/04/2014, Chandra Sekhar Divi <chandu_divi(a)hotmail.com> wrote:
Thanks a lot for the information.
Adding 'Thread.sleep(1);' after every insert reduces the problem intensity,
but sometimes even this solution also does not perform all the expected
actions (May be some events are dropped due to race condition).
Is there a way I can wait for KIE session to digest the event before I can
insert another event?
Not unless you consider inserting and firing all rules in a
synchronous (single-threaded) application. The feasibility depends on
the frequency of arriving events, but if you receive one temperature
reading per minute from each city or town in India (less than 1,700),
you won't have any problems.
FWIW, the race condition does not occur with Drools versions 5.3.0 and
5.5.0; so it should not be in 5.6 either.
-W
Thanks,
Chandra.
--
View this message in context:
http://drools.46999.n3.nabble.com/Drools-Fusion-Dropping-Actions-to-Event...
Sent from the Drools: User forum mailing list archive at Nabble.com.
_______________________________________________
rules-users mailing list
rules-users(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/rules-users
11:45 p.m.
Hi Wolfgang,
Thanks for the information, It works fine with the 5.6 version.
Thanks,
Chandra.
--
View this message in context:
http://drools.46999.n3.nabble.com/Drools-Fusion-Dropping-Actions-to-Event...
Sent from the Drools: User forum mailing list archive at Nabble.com.
7:21 a.m.
Hi,
I developed a test case almost identical to the one you suggested and I am
pasting it at the end of this email.
In the last months we did a few fixes on the event processing part and
indeed this test works as expected, i.e. the assertion at the end of the
test is true meaning that all the 10K inserted events get correctly
processed.
I suggest you to try this on the 6.1.0.Beta2 version that should already
contain all the fixes related to this use case. Anyway we will drop a Beta3
before the end of this week.
I hope this helps,
Mario
--------
@Test
public void testThroughput() throws Exception {
String drl = "package drools\n" +
"\n" +
"import " + TemperatureEvent.class.getCanonicalName() +
"\n" +
"global java.util.concurrent.atomic.AtomicInteger
counter\n" +
"\n" +
"declare TemperatureEvent\n" +
"@role(event)\n" +
"end\n" +
"\n" +
"rule \"RULE S1\"\n" +
"when\n" +
" $te:TemperatureEvent(measure <= 50) from
entry-point entryone\n" +
"then\n" +
" counter.incrementAndGet();\n" +
//" System.err.println(\"The temperature at location
- \"+$te.getLocation() + \" is \" + $te.getMeasure());\n" +
"end\n" +
"\n" +
"rule \"RULE S2\"\n" +
"when\n" +
" $te:TemperatureEvent(measure > 50) from
entry-point entryone\n" +
"then\n" +
" counter.incrementAndGet();\n" +
//" System.err.println(\"The temperature at location
- \"+$te.getLocation() + \" is \" + $te.getMeasure());\n" +
"end\n";
KieServices ks = KieServices.Factory.get();
KieFileSystem kfs = ks.newKieFileSystem();
KieModuleModel kieModule = ks.newKieModuleModel();
kieModule.newKieBaseModel("KBase")
.setDefault(true)
.setEventProcessingMode(EventProcessingOption.STREAM)
.newKieSessionModel("KSession")
.setDefault(true);
kfs.writeKModuleXML(kieModule.toXML());
kfs.write("src/main/resources/lifecycle.drl", drl);
KieBuilder builder = ks.newKieBuilder(kfs).buildAll();
assertEquals(0, builder.getResults().getMessages().size());
final KieSession kSession =
ks.newKieContainer(ks.getRepository().getDefaultReleaseId()).newKieSession();
AtomicInteger counter = new AtomicInteger(0);
kSession.setGlobal("counter", counter);
EntryPoint entryPoint1 = kSession.getEntryPoint("entryone");
new Thread() {
@Override
public void run() {
kSession.fireUntilHalt();
}
}.start();
Random measure = new Random(System.currentTimeMillis());
Random location = new Random(System.currentTimeMillis() - 1000000L);
String locations[] = { "Bangalore", "Chennai",
"Delhi", "Mumbai",
"Kolkata" };
for (int i = 0; i < 10; ++i) {
for (int j = 0; j < 1000; ++j) {
TemperatureEvent te = new TemperatureEvent(
measure.nextInt(100),
locations[location.nextInt(5)],
System.currentTimeMillis());
entryPoint1.insert(te);
}
}
Thread.sleep(100L);
kSession.halt();
kSession.dispose();
assertEquals(10000, counter.get());
}
public static class TemperatureEvent {
int measure;
String location;
long timeStamp;
public TemperatureEvent(int measure, String location, long
timeStamp) {
this.measure = measure;
this.location = location;
this.timeStamp = timeStamp;
}
public int getMeasure() {
return measure;
}
public void setMeasure(int measure) {
this.measure = measure;
}
public String getLocation() {
return location;
}
public void setLocation(String location) {
this.location = location;
}
public long getTimeStamp() {
return timeStamp;
}
public void setTimeStamp(long timeStamp) {
this.timeStamp = timeStamp;
}
}
--
View this message in context:
http://drools.46999.n3.nabble.com/Drools-Fusion-Dropping-Actions-to-Event...
Sent from the Drools: User forum mailing list archive at Nabble.com.
8:21 a.m.
Thanks Mario, It works with 6.1.0.Beta2.
-Chandra.
--
View this message in context:
http://drools.46999.n3.nabble.com/Drools-Fusion-Dropping-Actions-to-Event...
Sent from the Drools: User forum mailing list archive at Nabble.com.
3655
days inactive
3657
days old
6 comments
4 participants
participants (4)
-
Chandra Sekhar Divi -
chandu_divi
-
Mario Fusco -
Wolfgang Laun