Merge branch 'master' of https://github.com/RWTH-i5-IDSG/steve

JavaIsJavaScript · JavaIsJavaScript · commit bb34ffe19cf4 · 2018-04-19T13:07:50.000+02:00
diff --git a/src/main/java/de/rwth/idsg/steve/repository/impl/ChargePointRepositoryImpl.java b/src/main/java/de/rwth/idsg/steve/repository/impl/ChargePointRepositoryImpl.java
@@ -1,5 +1,6 @@
 package de.rwth.idsg.steve.repository.impl;
 
+import com.google.common.util.concurrent.Striped;
 import de.rwth.idsg.steve.SteveException;
 import de.rwth.idsg.steve.ocpp.OcppProtocol;
 import de.rwth.idsg.steve.ocpp.OcppTransport;
@@ -31,6 +32,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.locks.Lock;
 import java.util.stream.Collectors;
 
 import static de.rwth.idsg.steve.SteveConfiguration.CONFIG;
@@ -49,6 +51,8 @@
 public class ChargePointRepositoryImpl implements ChargePointRepository {
 
     private final boolean autoRegisterUnknownStations = CONFIG.getOcpp().isAutoRegisterUnknownStations();
+    private final Striped<Lock> isRegisteredLocks = Striped.lock(16);
+
     private final DSLContext ctx;
     private final AddressRepository addressRepository;
 
@@ -60,24 +64,30 @@ public ChargePointRepositoryImpl(DSLContext ctx, AddressRepository addressReposi
 
     @Override
     public boolean isRegistered(String chargeBoxId) {
-        // 1. exit if already registered
-        if (isRegisteredInternal(chargeBoxId)) {
-            return true;
-        }
+        Lock l = isRegisteredLocks.get(chargeBoxId);
+        l.lock();
+        try {
+            // 1. exit if already registered
+            if (isRegisteredInternal(chargeBoxId)) {
+                return true;
+            }
 
-        // 2. ok, this chargeBoxId is unknown. exit if auto-register is disabled
-        if (!autoRegisterUnknownStations) {
-            return false;
-        }
+            // 2. ok, this chargeBoxId is unknown. exit if auto-register is disabled
+            if (!autoRegisterUnknownStations) {
+                return false;
+            }
 
-        // 3. chargeBoxId is unknown and auto-register is enabled. insert chargeBoxId
-        try {
-            addChargePoint(Collections.singletonList(chargeBoxId));
-            log.warn("Auto-registered unknown chargebox '{}'", chargeBoxId);
-            return true;
-        } catch (Exception e) {
-            log.error("Failed to auto-register unknown chargebox '" + chargeBoxId + "'", e);
-            return false;
+            // 3. chargeBoxId is unknown and auto-register is enabled. insert chargeBoxId
+            try {
+                addChargePoint(Collections.singletonList(chargeBoxId));
+                log.warn("Auto-registered unknown chargebox '{}'", chargeBoxId);
+                return true;
+            } catch (Exception e) {
+                log.error("Failed to auto-register unknown chargebox '" + chargeBoxId + "'", e);
+                return false;
+            }
+        } finally {
+            l.unlock();
         }
     }
 
diff --git a/src/main/java/de/rwth/idsg/steve/service/UnidentifiedIncomingObjectService.java b/src/main/java/de/rwth/idsg/steve/service/UnidentifiedIncomingObjectService.java
@@ -3,9 +3,11 @@
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import de.rwth.idsg.steve.service.dto.UnidentifiedIncomingObject;
+import lombok.extern.slf4j.Slf4j;
 
 import java.util.Comparator;
 import java.util.List;
+import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 
 /**
@@ -14,6 +16,7 @@
  * @author Sevket Goekay <goekay@dbis.rwth-aachen.de>
  * @since 20.03.2018
  */
+@Slf4j
 public class UnidentifiedIncomingObjectService {
 
     private final Cache<String, UnidentifiedIncomingObject> objectsHolder;
@@ -33,13 +36,11 @@ public List<UnidentifiedIncomingObject> getObjects() {
     }
 
     public void processNewUnidentified(String key) {
-        synchronized (objectsHolder) {
-            UnidentifiedIncomingObject value = objectsHolder.getIfPresent(key);
-            if (value == null) {
-                objectsHolder.put(key, new UnidentifiedIncomingObject(key));
-            } else {
-                value.updateStats();
-            }
+        try {
+            objectsHolder.get(key, () -> new UnidentifiedIncomingObject(key))
+                         .updateStats();
+        } catch (ExecutionException e) {
+            log.error("Error occurred", e);
         }
     }
 }
diff --git a/src/main/java/de/rwth/idsg/steve/service/dto/UnidentifiedIncomingObject.java b/src/main/java/de/rwth/idsg/steve/service/dto/UnidentifiedIncomingObject.java
@@ -18,10 +18,9 @@ public class UnidentifiedIncomingObject {
 
     public UnidentifiedIncomingObject(String key) {
         this.key = key;
-        updateStats();
     }
 
-    public void updateStats() {
+    public synchronized void updateStats() {
         numberOfAttempts++;
         lastAttemptTimestamp = DateTime.now();
     }
diff --git a/src/test/java/de/rwth/idsg/steve/StressTestSoapOCPP16.java b/src/test/java/de/rwth/idsg/steve/StressTestSoapOCPP16.java
@@ -2,95 +2,206 @@
 
 import de.rwth.idsg.steve.utils.StressTester;
 import de.rwth.idsg.steve.utils.__DatabasePreparer__;
-import lombok.extern.slf4j.Slf4j;
 import ocpp.cs._2015._10.AuthorizationStatus;
 import ocpp.cs._2015._10.AuthorizeRequest;
 import ocpp.cs._2015._10.AuthorizeResponse;
 import ocpp.cs._2015._10.BootNotificationRequest;
 import ocpp.cs._2015._10.BootNotificationResponse;
 import ocpp.cs._2015._10.CentralSystemService;
+import ocpp.cs._2015._10.ChargePointErrorCode;
+import ocpp.cs._2015._10.ChargePointStatus;
+import ocpp.cs._2015._10.HeartbeatRequest;
+import ocpp.cs._2015._10.HeartbeatResponse;
+import ocpp.cs._2015._10.MeterValue;
+import ocpp.cs._2015._10.MeterValuesRequest;
+import ocpp.cs._2015._10.MeterValuesResponse;
 import ocpp.cs._2015._10.RegistrationStatus;
-import org.junit.After;
-import org.junit.AfterClass;
+import ocpp.cs._2015._10.SampledValue;
+import ocpp.cs._2015._10.StartTransactionRequest;
+import ocpp.cs._2015._10.StartTransactionResponse;
+import ocpp.cs._2015._10.StatusNotificationRequest;
+import ocpp.cs._2015._10.StatusNotificationResponse;
+import ocpp.cs._2015._10.StopTransactionRequest;
+import ocpp.cs._2015._10.StopTransactionResponse;
+import org.joda.time.DateTime;
 import org.junit.Assert;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
 
 import static de.rwth.idsg.steve.utils.Helpers.getForOcpp16;
 import static de.rwth.idsg.steve.utils.Helpers.getPath;
 import static de.rwth.idsg.steve.utils.Helpers.getRandomString;
+import static de.rwth.idsg.steve.utils.Helpers.getRandomStrings;
 
 /**
  * @author Sevket Goekay <goekay@dbis.rwth-aachen.de>
  * @since 18.04.2018
  */
-@Slf4j
 public class StressTestSoapOCPP16 {
 
-    private static final String REGISTERED_CHARGE_BOX_ID = __DatabasePreparer__.getRegisteredChargeBoxId();
+    // higher values -> more stress
+    //
+    private static final int THREAD_COUNT = 50;
+    private static final int REPEAT_COUNT_PER_THREAD = 50;
+
+    // lower values -> more stress
+    //
+    // reason: these only specify the size of the values "bag" from which a test picks a value randomly. if there is
+    // less values to pick from, it is more likely that tests will use the same value at the same time. this produces
+    // more overhead for steve (especially db) when multiple threads "fight" for inserting/updating a db row/cell.
+    //
+    private static final int ID_TAG_COUNT = 50;
+    private static final int CHARGE_BOX_COUNT = 100;
+    private static final int CONNECTOR_COUNT_PER_CHARGE_BOX = 25;
 
     private static final String path = getPath();
-    private static Application app;
 
-    @BeforeClass
-    public static void initClass() throws Exception {
+    public static void main(String[] args) throws Exception {
+        new StressTestSoapOCPP16().attack();
+    }
+
+    private void attack() throws Exception {
         Assert.assertEquals(ApplicationProfile.TEST, SteveConfiguration.CONFIG.getProfile());
         Assert.assertTrue(SteveConfiguration.CONFIG.getOcpp().isAutoRegisterUnknownStations());
 
-        app = new Application();
-        app.start();
-    }
+        __DatabasePreparer__.prepare();
 
-    @AfterClass
-    public static void destroyClass() throws Exception {
-        if (app != null) {
-            app.stop();
+        Application app = new Application();
+        try {
+            app.start();
+            attackInternal();
+        } finally {
+            try {
+                app.stop();
+            } finally {
+                __DatabasePreparer__.cleanUp();
+            }
         }
     }
 
-    @Before
-    public void init() {
-        __DatabasePreparer__.prepare();
-    }
+    private static void attackInternal() throws Exception {
+        final List<String> idTags = getRandomStrings(ID_TAG_COUNT);
+        final List<String> chargeBoxIds = getRandomStrings(CHARGE_BOX_COUNT);
 
-    @After
-    public void destroy() {
-        __DatabasePreparer__.cleanUp();
-    }
+        StressTester tester = new StressTester(THREAD_COUNT, REPEAT_COUNT_PER_THREAD);
 
-    @Test
-    public void testBootNotification() throws Exception {
-        StressTester tester = new StressTester(50, 5);
+        tester.test(() -> {
+            CentralSystemService client = getForOcpp16(path);
+            ThreadLocalRandom localRandom = ThreadLocalRandom.current();
 
-        CentralSystemService client = getForOcpp16(path);
+            String idTag = idTags.get(localRandom.nextInt(idTags.size()));
+            String chargeBoxId = chargeBoxIds.get(localRandom.nextInt(chargeBoxIds.size()));
+            int connectorId = localRandom.nextInt(1, CONNECTOR_COUNT_PER_CHARGE_BOX + 1);
 
-        tester.test(() -> {
+            int transactionStart = localRandom.nextInt(0, Integer.MAX_VALUE);
+            int transactionStop = localRandom.nextInt(transactionStart + 1, Integer.MAX_VALUE);
+
+            // to insert chargeBoxId into db
             BootNotificationResponse boot = client.bootNotification(
                     new BootNotificationRequest()
                             .withChargePointVendor(getRandomString())
                             .withChargePointModel(getRandomString()),
-                    getRandomString());
+                    chargeBoxId);
             Assert.assertEquals(RegistrationStatus.ACCEPTED, boot.getStatus());
+
+            HeartbeatResponse heartbeat = client.heartbeat(
+                    new HeartbeatRequest(),
+                    chargeBoxId
+            );
+            Assert.assertNotNull(heartbeat);
+
+            for (int i = 0; i <= CONNECTOR_COUNT_PER_CHARGE_BOX; i++) {
+                StatusNotificationResponse status = client.statusNotification(
+                        new StatusNotificationRequest()
+                                .withErrorCode(ChargePointErrorCode.NO_ERROR)
+                                .withStatus(ChargePointStatus.AVAILABLE)
+                                .withConnectorId(i)
+                                .withTimestamp(DateTime.now()),
+                        chargeBoxId
+                );
+                Assert.assertNotNull(status);
+            }
+
+            AuthorizeResponse auth = client.authorize(
+                    new AuthorizeRequest().withIdTag(idTag),
+                    chargeBoxId
+            );
+            Assert.assertNotEquals(AuthorizationStatus.ACCEPTED, auth.getIdTagInfo().getStatus());
+
+            StartTransactionResponse start = client.startTransaction(
+                    new StartTransactionRequest()
+                            .withConnectorId(connectorId)
+                            .withIdTag(idTag)
+                            .withTimestamp(DateTime.now())
+                            .withMeterStart(transactionStart),
+                    chargeBoxId
+            );
+            Assert.assertNotNull(start);
+
+            StatusNotificationResponse statusStart = client.statusNotification(
+                    new StatusNotificationRequest()
+                            .withErrorCode(ChargePointErrorCode.NO_ERROR)
+                            .withStatus(ChargePointStatus.CHARGING)
+                            .withConnectorId(connectorId)
+                            .withTimestamp(DateTime.now()),
+                    chargeBoxId
+            );
+            Assert.assertNotNull(statusStart);
+
+            MeterValuesResponse meter = client.meterValues(
+                    new MeterValuesRequest()
+                            .withConnectorId(connectorId)
+                            .withTransactionId(start.getTransactionId())
+                            .withMeterValue(getMeterValues(transactionStart, transactionStop)),
+                    chargeBoxId
+            );
+            Assert.assertNotNull(meter);
+
+            StopTransactionResponse stop = client.stopTransaction(
+                    new StopTransactionRequest()
+                            .withTransactionId(start.getTransactionId())
+                            .withTimestamp(DateTime.now())
+                            .withIdTag(idTag)
+                            .withMeterStop(transactionStop),
+                    chargeBoxId
+            );
+            Assert.assertNotNull(stop);
+
+            StatusNotificationResponse statusStop = client.statusNotification(
+                    new StatusNotificationRequest()
+                            .withErrorCode(ChargePointErrorCode.NO_ERROR)
+                            .withStatus(ChargePointStatus.AVAILABLE)
+                            .withConnectorId(connectorId)
+                            .withTimestamp(DateTime.now()),
+                    chargeBoxId
+            );
+            Assert.assertNotNull(statusStop);
         });
 
         tester.shutDown();
     }
 
-    @Test
-    public void testAuthorize() throws Exception {
-        StressTester tester = new StressTester(100, 10);
-
-        CentralSystemService client = getForOcpp16(path);
+    private static List<MeterValue> getMeterValues(int transactionStart, int transactionStop) {
+        final int size = 4;
+        int delta = (transactionStop - transactionStart) / size;
+        if (delta == 0) {
+            return Collections.emptyList();
+        }
 
-        tester.test(() -> {
-            AuthorizeResponse auth = client.authorize(
-                    new AuthorizeRequest().withIdTag(getRandomString()),
-                    REGISTERED_CHARGE_BOX_ID);
-            Assert.assertEquals(AuthorizationStatus.INVALID, auth.getIdTagInfo().getStatus());
-        });
+        List<MeterValue> list = new ArrayList<>(size);
+        for (int i = 0; i < size; i++) {
+            int meterValue = transactionStart + delta * (i + 1);
+            list.add(createMeterValue(meterValue));
+        }
+        return list;
+    }
 
-        tester.shutDown();
+    private static MeterValue createMeterValue(int val) {
+        return new MeterValue().withTimestamp(DateTime.now())
+                               .withSampledValue(new SampledValue().withValue(Integer.toString(val)));
     }
 
-}
+}
diff --git a/src/test/java/de/rwth/idsg/steve/utils/Helpers.java b/src/test/java/de/rwth/idsg/steve/utils/Helpers.java
@@ -4,6 +4,8 @@
 import org.apache.cxf.ws.addressing.WSAddressingFeature;
 
 import javax.xml.ws.soap.SOAPBinding;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.UUID;
 
 import static de.rwth.idsg.steve.SteveConfiguration.CONFIG;
@@ -18,6 +20,14 @@ public static String getRandomString() {
         return UUID.randomUUID().toString();
     }
 
+    public static List<String> getRandomStrings(int size) {
+        List<String> list = new ArrayList<>(size);
+        for (int i = 0; i < size; i++) {
+            list.add(getRandomString());
+        }
+        return list;
+    }
+
     public static String getPath() {
         if (CONFIG.getJetty().isHttpEnabled()) {
             return "http://"

Original file line number	Diff line number	Diff line change
`@@ -18,10 +18,9 @@ public class UnidentifiedIncomingObject {`
`18`	`18`
`19`	`19`	`public UnidentifiedIncomingObject(String key) {`
`20`	`20`	`this.key = key;`
`21`		`- updateStats();`
`22`	`21`	`}`
`23`	`22`
`24`		`- public void updateStats() {`
	`23`	`+ public synchronized void updateStats() {`
`25`	`24`	`numberOfAttempts++;`
`26`	`25`	`lastAttemptTimestamp = DateTime.now();`
`27`	`26`	`}`