/Users/nah/Documents/code/cast/src/java/cast/architecture/WorkingMemoryReaderComponent.java

Go to the documentation of this file.

/*
 * cast - the CoSy Architecture Schema Toolkit Copyright (C) 2006-2007
 * Nick Hawes This library is free software; you can redistribute it
 * and/or modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either version
 * 2.1 of the License, or (at your option) any later version. This
 * library is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 * Lesser General Public License for more details. You should have
 * received a copy of the GNU Lesser General Public License along with
 * this library; if not, write to the Free Software Foundation, Inc., 51
 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 */

package cast.architecture;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.Vector;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Semaphore;

import Ice.Current;
import cast.CASTException;
import cast.DoesNotExistOnWMException;
import cast.SubarchitectureComponentException;
import cast.UnknownSubarchitectureException;
import cast.cdl.FilterRestriction;
import cast.cdl.RECEIVERPRIORITYHIGH;
import cast.cdl.RECEIVERPRIORITYLOW;
import cast.cdl.RECEIVERPRIORITYMEDIUM;
import cast.cdl.WorkingMemoryAddress;
import cast.cdl.WorkingMemoryChange;
import cast.cdl.WorkingMemoryChangeFilter;
import cast.cdl.WorkingMemoryChangeQueueBehaviour;
import cast.cdl.WorkingMemoryEntry;
import cast.cdl.WorkingMemoryEntrySeqHolder;
import cast.core.CASTData;
import cast.core.CASTUtils;
import cast.core.ControlledRunnable;
import cast.core.SinglePlaceQueue;
import cast.core.logging.ComponentLogger;
import cast.interfaces.WorkingMemoryPrx;
import cast.interfaces.WorkingMemoryPrxHelper;
import cast.interfaces._WorkingMemoryReaderComponentOperations;

public abstract class WorkingMemoryReaderComponent extends
                WorkingMemoryWriterComponent implements
                _WorkingMemoryReaderComponentOperations {

        public enum ChangeReceiverPriority {
                HIGH(RECEIVERPRIORITYHIGH.value), MEDIUM(RECEIVERPRIORITYMEDIUM.value), LOW(
                                RECEIVERPRIORITYLOW.value);
                public int priority;

                private ChangeReceiverPriority(int _priority) {
                        priority = _priority;
                }
        }

        private class WorkingMemoryChangeThread extends ControlledRunnable {

                private Queue<WorkingMemoryChange> m_changes;

                private WorkingMemoryReaderComponent m_gdp;

                private Semaphore m_changeSemaphore;

                private final Queue<WorkingMemoryChangeReceiver> m_receivers;

                public WorkingMemoryChangeThread(WorkingMemoryReaderComponent _gdp) {
                        // m_changes = new
                        // ConcurrentLinkedQueue<WorkingMemoryChange>();
                        m_gdp = _gdp;

                        if (m_gdp.m_queueBehaviour == WorkingMemoryChangeQueueBehaviour.DISCARD) {
                                m_changes = new SinglePlaceQueue<WorkingMemoryChange>();
                        } else if (m_gdp.m_queueBehaviour == WorkingMemoryChangeQueueBehaviour.QUEUE) {
                                m_changes = new ConcurrentLinkedQueue<WorkingMemoryChange>();
                        }

                        m_changeSemaphore = new Semaphore(0, true);
                        m_receivers = new LinkedList<WorkingMemoryChangeReceiver>();
                }

                private boolean forwardToSubclass(WorkingMemoryChange[] changeList)
                                throws SubarchitectureComponentException {

                        boolean used = false;
                        try {
                                // if non null then we should generate event objects
                                if (m_gdp.m_changeObjects != null) {

                                        // remove any previously flagged filters
                                        removeChangeFilters();

                                        for (WorkingMemoryChange wmc : changeList) {
                                                // // get the object for the change
                                                // WorkingMemoryChangeReceiver receiver =
                                                // m_gdp.m_changeObjects
                                                // .get(new OLDWorkingMemoryChangeFilter(
                                                // wmc.m_type, wmc.m_operation));
                                                // if (receiver != null) {
                                                // receiver.workingMemoryChanged(wmc);
                                                // }
                                                // else {
                                                // throw new SubarchitectureComponentException(
                                                // "Missing change object for filtered change: "
                                                // + wmc.m_type + " " + wmc.m_operation);
                                                // }
                                                //
                                                // get the object for the change

                                                m_gdp.m_changeObjects.get(wmc, m_receivers);

                                                if (!m_receivers.isEmpty()) {

                                                        logSubscribedChange(wmc);

                                                        for (WorkingMemoryChangeReceiver receiver : m_receivers) {
                                                                try {
                                                                        receiver.workingMemoryChanged(wmc);
                                                                        used = true;
                                                                } catch (CASTException e) {
                                                                        m_logger.error("*********************************************************");
                                                                        m_logger.error("*********************************************************");
                                                                        m_logger.error("Exception thrown by change receiver: "
                                                                                        + e.message);
                                                                        m_logger.error("Change event: "
                                                                                        + CASTUtils.toString(wmc));
                                                                        logException(e);
                                                                        m_logger.error("*********************************************************");
                                                                        m_logger.error("*********************************************************");

                                                                } catch (Throwable e) {
                                                                        m_logger.error("*********************************************************");
                                                                        m_logger.error("*********************************************************");
                                                                        m_logger.error("Something unexpected thrown by change receiver, exiting");
                                                                        m_logger.error("Change event: "
                                                                                        + CASTUtils.toString(wmc));
                                                                        logException(e);
                                                                        m_logger.error("*********************************************************");
                                                                        m_logger.error("*********************************************************");
                                                                        System.exit(1);
                                                                }
                                                        }
                                                        m_receivers.clear();
                                                } else {
                                                        logUnsubscribedChange(wmc);
                                                }

                                                // remove filters flagged by that change
                                                removeChangeFilters();

                                        }

                                }
                        } catch (java.util.ConcurrentModificationException e) {
                                log(e.getStackTrace());
                                logException(e);
                                System.exit(0);
                                throw e;
                        }

                        // this looks like a good place to give up control
                        Thread.yield();

                        return used;
                }

                private final void removeChangeFilters()
                                throws SubarchitectureComponentException {
                        if (!m_gdp.m_receiversToRemove.isEmpty()) {
                                for (WorkingMemoryChangeReceiver receiver : m_gdp.m_receiversToRemove) {
                                        debug("removing receiver: " + receiver);
                                        m_gdp.removeChangeFilterHelper(receiver);
                                }
                                m_gdp.m_receiversToRemove.clear();
                        }
                }

                private void runDiscard() throws SubarchitectureComponentException {
                        WorkingMemoryChange[] changeList = new WorkingMemoryChange[1];
                        boolean used = false;
                        while (isRunning()) {

                                while (isRunning() && !m_changes.isEmpty()) {

                                        // try to gain access
                                        // if (m_gdp.m_semaphore.tryAcquire()) {
                                        m_gdp.lockComponent();

                                        // get changes from queue

                                        synchronized (m_changes) {
                                                if (m_changes.size() > 1) {
                                                        m_gdp.debug("discarding " + (m_changes.size() - 1)
                                                                        + " change events ");
                                                } else {
                                                        // m_gdp.log("forwarding 1 event");
                                                }

                                                changeList[0] = m_changes.poll();
                                                m_changes.clear();

                                        }

                                        try {
                                                // write to derived class
                                                used = forwardToSubclass(changeList);
                                        } catch (SubarchitectureComponentException e) {
                                                logException(e);
                                        } catch (RuntimeException e) {
                                                m_logger.error("Caught RuntimeException and rethrowing: "
                                                                + e.getMessage());
                                                throw e;
                                        }

                                        // UPGRADE now gone
                                        // synchronized (m_gdp.m_componentStatus) {
                                        // m_gdp.m_componentStatus.m_changeQueue = 0;
                                        // }

                                        // unlock derived class now something has
                                        // been
                                        // done
                                        m_gdp.unlockComponent();

                                        // cout<<m_pWMRP->getComponentID()<<":
                                        // "<<" RUN DONE "<<endl;

                                        // }
                                        // else {
                                        // m_gdp.log("unable to access thread");
                                        // synchronized (m_gdp.m_unlockNotification) {
                                        // while (isRunning()
                                        // && m_gdp.m_semaphore.availablePermits() == 0)
                                        // {
                                        // try {
                                        // m_gdp.m_unlockNotification.wait();
                                        // }
                                        // catch (InterruptedException e) {
                                        // logException(e);
                                        // System.exit(1);
                                        // }
                                        // }
                                        // }
                                        // }

                                }

                                if (used) {
                                        synchronized (m_gdp.m_wmChangeLock) {
                                                // now wake up any waiting threads
                                                m_gdp.m_wmChangeLock.notifyAll();
                                        }
                                }
                                try {
                                        m_changeSemaphore.acquire();
                                } catch (InterruptedException e) {
                                        logException(e);
                                        System.exit(1);
                                }

                        }
                }

                private void runQueue() throws SubarchitectureComponentException {

                        WorkingMemoryChange[] changeList;
                        boolean used = false;

                        while (isRunning()) {

                                while (!m_changes.isEmpty()) {

                                        // try to gain access
                                        // if (m_gdp.m_semaphore.tryAcquire()) {
                                        m_gdp.lockComponent();
                                        synchronized (m_changes) {
                                                // log(m_changes.size());
                                                changeList = new WorkingMemoryChange[m_changes.size()];
                                                // log(m_changes.size());
                                                for (int i = 0; i < changeList.length; i++) {
                                                        changeList[i] = m_changes.poll();
                                                }
                                                // log(m_changes.size());
                                                assert m_changes.size() == 0 : "emptied change list is not empty!";

                                                // if (changeList.length > 1) {
                                                // System.out.print("list longer than 1...
                                                // ");
                                                // }

                                        }
                                        try {
                                                // write to derived class
                                                used = forwardToSubclass(changeList);
                                        } catch (SubarchitectureComponentException e) {
                                                logException(e);
                                        } catch (RuntimeException e) {
                                                m_logger.error("Caught RuntimeException and rethrowing: "
                                                                + e.getMessage());
                                                throw e;
                                        }

                                        // if (changeList.length > 1) {
                                        // System.out.println("done");
                                        // }

                                        // UPGRADE now gone
                                        // synchronized (m_gdp.m_componentStatus) {
                                        // m_gdp.m_componentStatus.m_changeQueue = 0;
                                        // }

                                        // unlock derived class now something has
                                        // been
                                        // done
                                        m_gdp.unlockComponent();

                                        if (used) {
                                                synchronized (m_gdp.m_wmChangeLock) {
                                                        // now wake up any waiting threads
                                                        m_gdp.m_wmChangeLock.notifyAll();
                                                }
                                        }
                                        // cout<<m_pWMRP->getComponentID()<<":
                                        // "<<" RUN DONE "<<endl;

                                        // }
                                        // else {
                                        // m_gdp.println("unable to access thread");
                                        // synchronized (m_gdp.m_unlockNotification) {
                                        // while (isRunning()
                                        // && m_gdp.m_semaphore.availablePermits() == 0)
                                        // {
                                        // try {
                                        // m_gdp.m_unlockNotification.wait();
                                        // }
                                        // catch (InterruptedException e) {
                                        // logException(e);
                                        // System.exit(1);
                                        // }
                                        // }
                                        // }
                                        //
                                        // }
                                }
                                // synchronized (this) {
                                // // check we should be running before sleep
                                // while (isRunning() && m_changes.isEmpty()) {
                                //
                                // try {
                                // wait();
                                // }
                                // catch (InterruptedException e) {
                                // logException(e);
                                // System.exit(1);
                                // }
                                // }
                                // }

                                try {
                                        m_changeSemaphore.acquire();
                                } catch (InterruptedException e) {
                                        logException(e);
                                        System.exit(1);
                                }

                        }
                }

                public void queueChange(WorkingMemoryChange _change) {

                        // synchronized (this) {

                        synchronized (m_changes) {
                                m_changes.add(_change);
                        }

                        // UPGRADE now gone
                        // synchronized (m_gdp.m_componentStatus) {
                        // m_gdp.m_componentStatus.m_changeQueue++;
                        // }

                        m_changeSemaphore.release();

                        // notify();
                        // }

                }

                public void run() {
                        try {
                                if (m_gdp.m_queueBehaviour == WorkingMemoryChangeQueueBehaviour.DISCARD) {
                                        m_changes = new SinglePlaceQueue<WorkingMemoryChange>();
                                        runDiscard();
                                } else if (m_gdp.m_queueBehaviour == WorkingMemoryChangeQueueBehaviour.QUEUE) {
                                        m_changes = new ConcurrentLinkedQueue<WorkingMemoryChange>();
                                        runQueue();
                                }
                        } catch (SubarchitectureComponentException e) {
                                logException(e);
                                System.exit(1);
                        }
                }

                public void stop() {

                        super.stop();

                        synchronized (m_changes) {
                                m_changes.clear();
                        }

                        // synchronized (this) {
                        // notify();
                        // }

                        m_changeSemaphore.release();
                }

        }

        private Object m_wmChangeLock;

        private WorkingMemoryChangeThread m_wmChangeRunnable;

        private Thread m_wmChangeThread;

        private Vector<WorkingMemoryChangeReceiver> m_receiversToRemove;

        protected boolean m_bReceivingChanges;

        protected boolean m_makeCopyOnRead;

        private WorkingMemoryChangeFilterMap<WorkingMemoryChangeReceiver> m_changeObjects;

        //
        // /**
        // * Flag to determine whether this component should receive change events
        // * from other subarchitectures.
        // */
        // protected boolean m_receiveXarchChangeNotifications;

        protected WorkingMemoryChangeQueueBehaviour m_queueBehaviour;

        private ComponentLogger m_loggerForGets;
        private ComponentLogger m_loggerForUnsubscribedChanges;
        private ComponentLogger m_loggerForSubscribedChanges;

        private WorkingMemoryPrx m_workingMemoryForRead;

        public WorkingMemoryReaderComponent() {
                receiveChanges();

                m_queueBehaviour = WorkingMemoryChangeQueueBehaviour.QUEUE;
                m_wmChangeRunnable = new WorkingMemoryChangeThread(this);
                m_wmChangeLock = new Object();

                m_changeObjects = new WorkingMemoryChangeFilterMap<WorkingMemoryChangeReceiver>();
                m_receiversToRemove = new Vector<WorkingMemoryChangeReceiver>();

                m_makeCopyOnRead = false;
        }

        // /**
        // * @param _filteredCount
        // * @param _totalCount
        // */
        // private void logChangeEvents(int _filteredCount, int _totalCount) {
        // synchronized (m_componentStatus) {
        // m_componentStatus.m_totalChangeEventsFiltered += _filteredCount;
        // m_componentStatus.m_totalChangeEventsReceived += _totalCount;
        // }
        // }
        //
        // private void logReadEvent(String _subarch, String _id) {
        // logEvent(ComponentEventType.GET, getComponentID(),
        // _subarch, "", _id);
        // }
        //
        // private void logReadEvent(String _subarch, String _type, int _count) {
        // logEvent(ComponentEventType.GET, getComponentID(),
        // _subarch, _type, "");
        // }

        protected int getFilterCount() {
                return m_changeObjects.size();
        }

        private void removeChangeFilterHelper(
                        WorkingMemoryChangeReceiver _workingMemoryChangeReceiver)
                        throws SubarchitectureComponentException {

                Vector<WorkingMemoryChangeFilter> removed = new Vector<WorkingMemoryChangeFilter>();
                m_changeObjects.remove(_workingMemoryChangeReceiver, removed);

                for (WorkingMemoryChangeFilter filter : removed) {
                        m_workingMemory.removeComponentFilter(filter);
                }

        }

        public void addChangeFilter(WorkingMemoryChangeFilter _filter,
                        WorkingMemoryChangeReceiver _receiver) {
                addChangeFilter(_filter, _receiver, ChangeReceiverPriority.MEDIUM);
        }

        public void addChangeFilter(WorkingMemoryChangeFilter _filter,
                        WorkingMemoryChangeReceiver _receiver,
                        ChangeReceiverPriority _priority) {
                addChangeFilter(_filter, _receiver, _priority.priority);
        }

        public void addChangeFilter(WorkingMemoryChangeFilter _filter,
                        WorkingMemoryChangeReceiver _receiver, int _priority) {

                assert (m_workingMemory != null);

                // sneaky operation to keep filter consistent without always
                // needing a component for info
                if (_filter.restriction == FilterRestriction.LOCALSA
                                && _filter.address.subarchitecture.length() == 0) {
                        _filter.address.subarchitecture = getSubarchitectureID();
                }
                // also make sure subarch/restrictions are correct
                else if (_filter.address.subarchitecture.equals(getSubarchitectureID())
                                && _filter.restriction == FilterRestriction.ALLSA) {
                        _filter.restriction = FilterRestriction.LOCALSA;
                }

                // and set origin
                _filter.origin = getComponentID();

                // if (existingReceiver == null) {

                m_changeObjects.put(_filter, _receiver, _priority);

                // println("new filters length: " + m_changeObjects.size());

                m_workingMemory.registerComponentFilter(_filter, _priority);
        }

        void getBaseWorkingMemoryEntries(String _type,
                        List<WorkingMemoryEntry> _entries, String _subarch, int _count)
                        throws UnknownSubarchitectureException {
                assert (_subarch.length() != 0);// subarch must not be empty
                assert (_type.length() != 0);
                assert (_entries != null);

                WorkingMemoryEntrySeqHolder holder = new WorkingMemoryEntrySeqHolder();
                m_workingMemoryForRead.getWorkingMemoryEntries(_type, _subarch, _count,
                                getComponentID(), holder);

                // update newly read things
                for (WorkingMemoryEntry entry : holder.value) {
                        updateVersion(entry.id, entry.version);
                        logGet(entry.id, _subarch, entry.type, entry.version);
                        _entries.add(entry);
                }
        }

        void getBaseWorkingMemoryEntries(String _type,
                        List<WorkingMemoryEntry> _entries, String _subarch)
                        throws UnknownSubarchitectureException {
                getBaseWorkingMemoryEntries(_type, _entries, _subarch, 0);
        }

        void getBaseWorkingMemoryEntries(String _type,
                        List<WorkingMemoryEntry> _entries, int _count) {
                try {
                        getBaseWorkingMemoryEntries(_type, _entries,
                                        getSubarchitectureID(), _count);
                } catch (UnknownSubarchitectureException e) {
                        throw new RuntimeException(
                                        "Shouldn't happen on own subarchitecture", e);
                }
        }

        public <Type extends Ice.Object> CASTData<Type>[] getWorkingMemoryEntries(
                        Class<Type> _cls, int _count)
                        throws SubarchitectureComponentException {
                return getWorkingMemoryEntries(getSubarchitectureID(), _cls, _count);
        }

        public <Type extends Ice.Object> CASTData<Type>[] getWorkingMemoryEntries(
                        Class<Type> _cls) throws SubarchitectureComponentException {
                return getWorkingMemoryEntries(_cls, 0);
        }

        public <Type extends Ice.Object> CASTData<Type>[] getWorkingMemoryEntries(
                        String _subarch, Class<Type> _cls)
                        throws SubarchitectureComponentException {

                return getWorkingMemoryEntries(_subarch, _cls, 0);
        }

        @SuppressWarnings("unchecked")
        @Deprecated
        public <Type extends Ice.Object> CASTData<Type>[] getWorkingMemoryEntries(
                        String _subarch, Class<Type> _cls, int _count)
                        throws UnknownSubarchitectureException {
                String type = CASTUtils.typeName(_cls);
                List<WorkingMemoryEntry> entries = new ArrayList<WorkingMemoryEntry>();
                getBaseWorkingMemoryEntries(type, entries, _subarch, _count);
                CASTData<Type>[] array = new CASTData[entries.size()];

                int i = 0;
                for (WorkingMemoryEntry entry : entries) {
                        array[i++] = new CASTData<Type>(entry, _cls);
                }

                return array;
        }

        public <Type extends Ice.Object> void getMemoryEntries(Class<Type> _cls,
                        List<Type> _entries, String _subarch, int _count)
                        throws UnknownSubarchitectureException {
                String type = CASTUtils.typeName(_cls);
                List<WorkingMemoryEntry> entries = new ArrayList<WorkingMemoryEntry>();
                getBaseWorkingMemoryEntries(type, entries, _subarch, _count);
                for (WorkingMemoryEntry entry : entries) {
                        _entries.add(_cls.cast(entry.entry));
                }
        }

        public <Type extends Ice.Object> void getMemoryEntries(Class<Type> _cls,
                        List<Type> _entries) throws UnknownSubarchitectureException {
                getMemoryEntries(_cls, _entries, getSubarchitectureID(), 0);
        }

        public <Type extends Ice.Object> void getMemoryEntries(Class<Type> _cls,
                        List<Type> _entries, String _subarch)
                        throws UnknownSubarchitectureException {
                getMemoryEntries(_cls, _entries, _subarch, 0);
        }

        public <Type extends Ice.Object> void getMemoryEntries(Class<Type> _cls,
                        List<Type> _entries, int _count) {
                try {
                        getMemoryEntries(_cls, _entries, getSubarchitectureID(), _count);
                } catch (UnknownSubarchitectureException e) {
                        throw new RuntimeException(
                                        "Shouldn't happen on own subarchitecture", e);
                }
        }

        public <Type extends Ice.Object> void getMemoryEntriesWithData(
                        Class<Type> _cls, List<CASTData<Type>> _entries, String _subarch,
                        int _count) throws UnknownSubarchitectureException {
                String type = CASTUtils.typeName(_cls);
                List<WorkingMemoryEntry> entries = new ArrayList<WorkingMemoryEntry>();
                getBaseWorkingMemoryEntries(type, entries, _subarch, _count);
                for (WorkingMemoryEntry entry : entries) {
                        _entries.add(new CASTData<Type>(entry, _cls));
                }
        }

        public <Type extends Ice.Object> void getMemoryEntriesWithDataWithData(
                        Class<Type> _cls, List<CASTData<Type>> _entries, String _subarch)
                        throws UnknownSubarchitectureException {
                getMemoryEntriesWithData(_cls, _entries, _subarch, 0);
        }

        public <Type extends Ice.Object> void getMemoryEntriesWithData(
                        Class<Type> _cls, List<CASTData<Type>> _entries, int _count) {
                try {
                        getMemoryEntriesWithData(_cls, _entries, getSubarchitectureID(),
                                        _count);
                } catch (UnknownSubarchitectureException e) {
                        throw new RuntimeException(
                                        "Shouldn't happen on own subarchitecture", e);
                }
        }

        WorkingMemoryEntry getBaseMemoryEntry(String _id)
                        throws DoesNotExistOnWMException {
                try {
                        return getBaseMemoryEntry(_id, getSubarchitectureID());
                } catch (UnknownSubarchitectureException e) {
                        throw new RuntimeException(
                                        "Shouldn't happen on own subarchitecture", e);
                }
        }

        WorkingMemoryEntry getBaseMemoryEntry(WorkingMemoryAddress _wma)
                        throws DoesNotExistOnWMException, UnknownSubarchitectureException {
                return getBaseMemoryEntry(_wma.id, _wma.subarchitecture);
        }

        WorkingMemoryEntry getBaseMemoryEntry(String _id, String _subarch)
                        throws DoesNotExistOnWMException, UnknownSubarchitectureException {

                assert (_id != null);
                assert (_subarch != null);
                assert (_id.length() != 0);
                assert (_subarch.length() != 0);
                assert (m_workingMemory != null);
                WorkingMemoryEntry entry = m_workingMemoryForRead
                                .getWorkingMemoryEntry(_id, _subarch, getComponentID());
                assert (entry != null);

                updateVersion(entry.id, entry.version);
                logGet(_id, _subarch, entry.type, entry.version);
                return entry;
        }

        @Deprecated
        public CASTData<?> getWorkingMemoryEntry(String _id)
                        throws DoesNotExistOnWMException {
                try {
                        return getWorkingMemoryEntry(_id, getSubarchitectureID());
                } catch (UnknownSubarchitectureException e) {
                        throw new RuntimeException(
                                        "Shouldn't happen on own subarchitecture", e);
                }
        }

        @SuppressWarnings("unchecked")
        @Deprecated
        public CASTData<?> getWorkingMemoryEntry(String _id, String _subarch)
                        throws DoesNotExistOnWMException, UnknownSubarchitectureException {
                WorkingMemoryEntry entry = getBaseMemoryEntry(_id, _subarch);
                return new CASTData(entry.id, entry.type, entry.version, entry.entry);
        }

        @Deprecated
        public CASTData<?> getWorkingMemoryEntry(WorkingMemoryAddress _wma)
                        throws SubarchitectureComponentException {
                return getWorkingMemoryEntry(_wma.id, _wma.subarchitecture);
        }

        public <T extends Ice.Object> T getMemoryEntry(String _id, Class<T> _cls)
                        throws DoesNotExistOnWMException {
                try {
                        return getMemoryEntry(_id, getSubarchitectureID(), _cls);
                } catch (UnknownSubarchitectureException e) {
                        throw new RuntimeException(
                                        "Shouldn't happen on own subarchitecture", e);
                }
        }

        public <T extends Ice.Object> T getMemoryEntry(String _id, String _subarch,
                        Class<T> _cls) throws DoesNotExistOnWMException,
                        UnknownSubarchitectureException {
                WorkingMemoryEntry entry = getBaseMemoryEntry(_id, _subarch);
                return _cls.cast(entry.entry);
        }

        public <T extends Ice.Object> T getMemoryEntry(WorkingMemoryAddress _wma,
                        Class<T> _cls) throws DoesNotExistOnWMException,
                        UnknownSubarchitectureException {
                return getMemoryEntry(_wma.id, _wma.subarchitecture, _cls);
        }

        public <T extends Ice.Object> CASTData<T> getMemoryEntryWithData(
                        String _id, Class<T> _cls) throws DoesNotExistOnWMException {
                try {
                        return getMemoryEntryWithData(_id, getSubarchitectureID(), _cls);
                } catch (UnknownSubarchitectureException e) {
                        throw new RuntimeException(
                                        "Shouldn't happen on own subarchitecture", e);
                }
        }

        public <T extends Ice.Object> CASTData<T> getMemoryEntryWithData(
                        String _id, String _subarch, Class<T> _cls)
                        throws DoesNotExistOnWMException, UnknownSubarchitectureException {
                WorkingMemoryEntry entry = getBaseMemoryEntry(_id, _subarch);
                return new CASTData<T>(entry, _cls);
        }

        public <T extends Ice.Object> CASTData<T> getMemoryEntryWithData(
                        WorkingMemoryAddress _wma, Class<T> _cls)
                        throws DoesNotExistOnWMException, UnknownSubarchitectureException {
                return getMemoryEntryWithData(_wma.id, _wma.subarchitecture, _cls);
        }

        protected void receiveChanges() {
                m_bReceivingChanges = true;
        }

        protected void receiveNoChanges() {
                m_bReceivingChanges = false;
        }

        public void removeChangeFilter(
                        WorkingMemoryChangeReceiver _workingMemoryChangeReceiver)
                        throws SubarchitectureComponentException {
                m_receiversToRemove.add(_workingMemoryChangeReceiver);
        }

        public void waitForChanges() {
                synchronized (m_wmChangeLock) {
                        try {
                                m_wmChangeLock.wait();
                        } catch (InterruptedException e) {
                                logException(e);
                                System.exit(1);
                        }
                }
        }

        public void receiveChangeEvent(WorkingMemoryChange _wmc, Current __current) {

                // no thread, and miss updates if blocking
                if (isRunning() && m_bReceivingChanges) {

                        // prefer to use change objects
                        if (m_changeObjects != null) {
                                m_wmChangeRunnable.queueChange(_wmc);
                        }
                }
        }

        @Override
        public void startInternal() {
                super.startInternal();

                m_wmChangeRunnable.start();
                m_wmChangeThread = new Thread(m_wmChangeRunnable, getComponentID()
                                + "-receiver");

                m_wmChangeThread.setPriority(Thread.NORM_PRIORITY + 1);
                m_wmChangeThread.start();

                m_loggerForGets = getLogger(".wm.rw.get");
                m_loggerForSubscribedChanges = getLogger(".wm.ch.sub");
                m_loggerForUnsubscribedChanges = getLogger(".wm.ch.all");
        }

        private final void logGet(String _id, String _subarch, String _type,
                        int _version) {

                if (m_loggerForGets.isTraceEnabled()) {
                        m_loggerForGets.trace(CASTUtils.concatenate("get,",
                                        CASTUtils.toString(getCASTTime()), ",", _id, ",", _subarch,
                                        ",", _type, ",", _version), getLogAdditions());
                }
        }

        protected void logChange(WorkingMemoryChange _wmc, ComponentLogger _logger) {
                _logger.trace(CASTUtils.concatenate("wmc,",
                                CASTUtils.toString(getCASTTime()), ",", _wmc.operation, ",",
                                _wmc.address.id, ",", _wmc.address.subarchitecture, ",",
                                _wmc.type), getLogAdditions());
        }

        private void logUnsubscribedChange(WorkingMemoryChange _wmc) {
                if (m_loggerForUnsubscribedChanges.isTraceEnabled()) {
                        logChange(_wmc, m_loggerForUnsubscribedChanges);
                }
        }

        private void logSubscribedChange(WorkingMemoryChange _wmc) {
                if (m_loggerForSubscribedChanges.isTraceEnabled()) {
                        logChange(_wmc, m_loggerForSubscribedChanges);
                }
        }

        @Override
        public void stopInternal() {
                super.stopInternal();

                // stop change thread
                m_wmChangeRunnable.stop();
                try {
                        m_wmChangeThread.join();
                } catch (InterruptedException e) {
                        logException(e);
                }

                // wake threads waiting on changes
                synchronized (m_wmChangeLock) {
                        m_wmChangeLock.notifyAll();
                }
        }

        @Override
        public void setWorkingMemory(WorkingMemoryPrx _wm, Current __current) {
                super.setWorkingMemory(_wm, __current);

                // Check whether the WM is language and network local
                if (m_workingMemory.ice_isCollocationOptimized()) {
                        // if it is, create a new WM proxy that forces serialisation to
                        // ensure a true deep copy is created
                        m_workingMemoryForRead = WorkingMemoryPrxHelper
                                        .uncheckedCast(m_workingMemory
                                                        .ice_collocationOptimized(false));
                } else {
                        m_workingMemoryForRead = m_workingMemory;
                }
        }

        protected boolean resetReadCollocationOptimisation() {
                m_workingMemoryForRead = m_workingMemory;
                return m_workingMemoryForRead.ice_isCollocationOptimized();
        }

        protected void turnOffReadCollocationOptimisation() {
                // Check whether the WM is language and network local
                if (m_workingMemoryForRead.ice_isCollocationOptimized()) {
                        m_workingMemoryForRead = WorkingMemoryPrxHelper
                                        .uncheckedCast(m_workingMemory
                                                        .ice_collocationOptimized(false));
                }
        }

}

---