server/src/test/java/com/cloud/network/security/SecurityGroupQueueTest.java

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.
package com.cloud.network.security;

import com.cloud.utils.Profiler;
import junit.framework.TestCase;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

public class SecurityGroupQueueTest extends TestCase {
    public final static SecurityGroupWorkQueue queue = new LocalSecurityGroupWorkQueue();

    public static class Producer implements Runnable {
        int _maxVmId = 0;
        int _newWorkQueued = 0;
        Set<Long> vmIds = new HashSet<Long>();

        public Producer(int maxVmId) {
            this._maxVmId = maxVmId;
            for (long i = 1; i <= _maxVmId; i++) {
                vmIds.add(i);
            }
        }

        @Override
        public void run() {
            _newWorkQueued = queue.submitWorkForVms(vmIds);
        }

        public int getNewWork() {
            return _newWorkQueued;
        }

        public int getTotalWork() {
            return _maxVmId;
        }
    }

    public static class Consumer implements Runnable {
        private int _numJobsToDequeue = 0;
        private int _numJobsDequeued = 0;

        public Consumer(int numJobsToDequeu) {
            this._numJobsToDequeue = numJobsToDequeu;
        }

        @Override
        public void run() {
            List<SecurityGroupWork> result = new ArrayList<SecurityGroupWork>();
            try {
                result = queue.getWork(_numJobsToDequeue);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            this._numJobsDequeued = result.size();
        }

        int getNumJobsToDequeue() {
            return _numJobsToDequeue;
        }

        int getNumJobsDequeued() {
            return _numJobsDequeued;
        }
    }

    public void testNumJobsEqToNumVms1() {
        queue.clear();
        final int numProducers = 50;
        Thread[] pThreads = new Thread[numProducers];

        Producer[] producers = new Producer[numProducers];
        int numProduced = 0;

        for (int i = 0; i < numProducers; i++) {
            producers[i] = new Producer(i + 1);
            pThreads[i] = new Thread(producers[i]);
            numProduced += i + 1;
            pThreads[i].start();
        }
        for (int i = 0; i < numProducers; i++) {
            try {
                pThreads[i].join();
            } catch (InterruptedException ie) {
                ie.printStackTrace();
            }
        }
        System.out.println("Num Vms= " + numProducers + " Queue size = " + queue.size());
        assertEquals(numProducers, queue.size());
    }

    protected void testNumJobsEqToNumVms2(int numProducers, int maxVmId) {
        queue.clear();

        Thread[] pThreads = new Thread[numProducers];

        Producer[] producers = new Producer[numProducers];
        int numProduced = 0;
        Profiler p = new Profiler();
        p.start();
        for (int i = 0; i < numProducers; i++) {
            producers[i] = new Producer(maxVmId);
            pThreads[i] = new Thread(producers[i]);
            numProduced += i + 1;
            pThreads[i].start();
        }
        for (int i = 0; i < numProducers; i++) {
            try {
                pThreads[i].join();
            } catch (InterruptedException ie) {
                ie.printStackTrace();
            }
        }
        p.stop();
        System.out.println("Num Vms= " + maxVmId + " Queue size = " + queue.size() + " time=" + p.getDurationInMillis() + " ms");
        assertEquals(maxVmId, queue.size());
    }

    public void testNumJobsEqToNumVms3() {
        testNumJobsEqToNumVms2(50, 20000);
        testNumJobsEqToNumVms2(400, 5000);
        testNumJobsEqToNumVms2(1, 1);
        testNumJobsEqToNumVms2(1, 1000000);
        testNumJobsEqToNumVms2(750, 1);

    }

    protected void _testDequeueOneJob(final int numConsumers, final int numProducers, final int maxVmId) {
        queue.clear();

        Thread[] pThreads = new Thread[numProducers];
        Thread[] cThreads = new Thread[numConsumers];

        Consumer[] consumers = new Consumer[numConsumers];
        Producer[] producers = new Producer[numProducers];

        int numProduced = 0;
        for (int i = 0; i < numConsumers; i++) {
            consumers[i] = new Consumer(1);
            cThreads[i] = new Thread(consumers[i]);
            cThreads[i].start();
        }
        for (int i = 0; i < numProducers; i++) {
            producers[i] = new Producer(maxVmId);
            pThreads[i] = new Thread(producers[i]);
            numProduced += maxVmId;
            pThreads[i].start();
        }
        for (int i = 0; i < numConsumers; i++) {
            try {
                cThreads[i].join();
            } catch (InterruptedException ie) {
                ie.printStackTrace();
            }
        }
        for (int i = 0; i < numProducers; i++) {
            try {
                pThreads[i].join();
            } catch (InterruptedException ie) {
                ie.printStackTrace();
            }
        }

        int totalDequeued = 0;
        for (int i = 0; i < numConsumers; i++) {
            //System.out.println("Consumer " + i + " ask to dequeue " + consumers[i].getNumJobsToDequeue() + ", dequeued " + consumers[i].getNumJobsDequeued());
            totalDequeued += consumers[i].getNumJobsDequeued();
        }
        int totalQueued = 0;
        for (int i = 0; i < numProducers; i++) {
            //System.out.println("Producer " + i + " ask to queue " + producers[i].getTotalWork() + ", queued " + producers[i].getNewWork());
            totalQueued += producers[i].getNewWork();
        }
        System.out.println("Total jobs dequeued = " + totalDequeued + ", num queued=" + totalQueued + " queue current size=" + queue.size());
        assertEquals(totalDequeued, numConsumers);
        assertEquals(totalQueued - totalDequeued, queue.size());
    }

    public void testDequeueOneJobAgain() {
        _testDequeueOneJob(10, 10, 1000);
        int queueSize = queue.size();
        Thread cThread = new Thread(new Consumer(1));
        cThread.start();
        try {
            cThread.join();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        assertEquals(queue.size(), queueSize - 1);
    }

    public void testDequeueOneJob() {
        _testDequeueOneJob(10, 10, 1000);
        _testDequeueOneJob(1, 10, 1000);
        _testDequeueOneJob(10, 1, 1000);
        _testDequeueOneJob(10, 1, 10);
    }

}