diff options
13 files changed, 1341 insertions, 20 deletions
diff --git a/plugins/org.eclipse.tm.tcf.cdt.ui/plugin.properties b/plugins/org.eclipse.tm.tcf.cdt.ui/plugin.properties index a2e03adb8..03735e665 100644 --- a/plugins/org.eclipse.tm.tcf.cdt.ui/plugin.properties +++ b/plugins/org.eclipse.tm.tcf.cdt.ui/plugin.properties @@ -39,3 +39,5 @@ command.addWatchpoint.desc=Allows to add a new watchpoint on an arbitrary symbol item.addFunctionBreakpoint.name=Add Function Breakpoint (C/C++)... item.addWatchpoint.name=Add Watchpoint... + +editor.sourceNotFound.name=Source Not Found diff --git a/plugins/org.eclipse.tm.tcf.cdt.ui/plugin.xml b/plugins/org.eclipse.tm.tcf.cdt.ui/plugin.xml index 62a22b176..71977144f 100644 --- a/plugins/org.eclipse.tm.tcf.cdt.ui/plugin.xml +++ b/plugins/org.eclipse.tm.tcf.cdt.ui/plugin.xml @@ -20,14 +20,8 @@ point="org.eclipse.core.runtime.adapters"> <factory adaptableType="org.eclipse.tm.internal.tcf.debug.ui.model.TCFNode" - class="org.eclipse.tm.internal.tcf.cdt.ui.disassembly.TCFDisassemblyBackendFactory"> - <adapter - type="org.eclipse.cdt.debug.internal.ui.disassembly.dsf.IDisassemblyBackend"> - </adapter> - </factory> - <factory - adaptableType="org.eclipse.tm.internal.tcf.debug.ui.model.TCFNode" class="org.eclipse.tm.internal.tcf.cdt.ui.TCFNodeAdapterFactory"> + <adapter type="org.eclipse.cdt.debug.internal.ui.disassembly.dsf.IDisassemblyBackend"/> <adapter type="org.eclipse.cdt.debug.core.model.ISteppingModeTarget"/> <adapter type="org.eclipse.debug.core.model.ISuspendResume"/> <adapter type="org.eclipse.cdt.ui.text.c.hover.ICEditorTextHover"/> @@ -37,6 +31,7 @@ <adapter type="org.eclipse.cdt.debug.core.model.IReverseResumeHandler"/> <adapter type="org.eclipse.cdt.debug.core.model.IUncallHandler"/> <adapter type="org.eclipse.cdt.debug.internal.core.ICWatchpointTarget"/> + <adapter type="org.eclipse.tm.internal.tcf.debug.ui.model.ISourceNotFoundPresentation"/> </factory> </extension> @@ -338,5 +333,15 @@ menubarPath="additions"/> </objectContribution> </extension> + <extension + point="org.eclipse.ui.editors"> + <editor + class="org.eclipse.tm.internal.tcf.cdt.ui.sourcelookup.TCFCSourceNotFoundEditor" + default="false" + icon="icons/obj16/c_app.gif" + id="org.eclipse.tm.tcf.cdt.ui.source_not_found_editor" + name="%editor.sourceNotFound.name"> + </editor> + </extension> </plugin> diff --git a/plugins/org.eclipse.tm.tcf.cdt.ui/src/org/eclipse/tm/internal/tcf/cdt/ui/TCFNodeAdapterFactory.java b/plugins/org.eclipse.tm.tcf.cdt.ui/src/org/eclipse/tm/internal/tcf/cdt/ui/TCFNodeAdapterFactory.java index d5fcb0ad4..a050ccbcc 100644 --- a/plugins/org.eclipse.tm.tcf.cdt.ui/src/org/eclipse/tm/internal/tcf/cdt/ui/TCFNodeAdapterFactory.java +++ b/plugins/org.eclipse.tm.tcf.cdt.ui/src/org/eclipse/tm/internal/tcf/cdt/ui/TCFNodeAdapterFactory.java @@ -17,6 +17,7 @@ import org.eclipse.cdt.debug.core.model.IReverseToggleHandler; import org.eclipse.cdt.debug.core.model.ISteppingModeTarget;
import org.eclipse.cdt.debug.core.model.IUncallHandler;
import org.eclipse.cdt.debug.internal.core.ICWatchpointTarget;
+import org.eclipse.cdt.debug.internal.ui.disassembly.dsf.IDisassemblyBackend;
import org.eclipse.cdt.ui.text.c.hover.ICEditorTextHover;
import org.eclipse.core.runtime.IAdapterFactory;
import org.eclipse.debug.core.model.ISuspendResume;
@@ -26,7 +27,10 @@ import org.eclipse.tm.internal.tcf.cdt.ui.commands.TCFReverseStepIntoCommand; import org.eclipse.tm.internal.tcf.cdt.ui.commands.TCFReverseStepOverCommand;
import org.eclipse.tm.internal.tcf.cdt.ui.commands.TCFReverseStepReturnCommand;
import org.eclipse.tm.internal.tcf.cdt.ui.commands.TCFReverseToggleCommand;
+import org.eclipse.tm.internal.tcf.cdt.ui.disassembly.TCFDisassemblyBackend;
import org.eclipse.tm.internal.tcf.cdt.ui.hover.TCFDebugTextHover;
+import org.eclipse.tm.internal.tcf.cdt.ui.sourcelookup.TCFSourceNotFoundPresentation;
+import org.eclipse.tm.internal.tcf.debug.ui.model.ISourceNotFoundPresentation;
import org.eclipse.tm.internal.tcf.debug.ui.model.TCFModel;
import org.eclipse.tm.internal.tcf.debug.ui.model.TCFNode;
import org.eclipse.tm.internal.tcf.debug.ui.model.TCFNodeExecContext;
@@ -37,7 +41,8 @@ import org.eclipse.tm.tcf.util.TCFTask; @SuppressWarnings({ "rawtypes", "restriction" })
public class TCFNodeAdapterFactory implements IAdapterFactory {
- private static final Class<?>[] CLASSES = {
+ private static final Class<?>[] CLASSES = {
+ IDisassemblyBackend.class,
ISteppingModeTarget.class,
ISuspendResume.class,
ICEditorTextHover.class,
@@ -48,12 +53,19 @@ public class TCFNodeAdapterFactory implements IAdapterFactory { IUncallHandler.class,
ICWatchpointTarget.class
};
+
+ private static final TCFSourceNotFoundPresentation fgSourceNotFoundPresentation = new TCFSourceNotFoundPresentation();
public Object getAdapter(Object adaptableObject, Class adapterType) {
if (adaptableObject instanceof TCFNode) {
final TCFNode node = (TCFNode) adaptableObject;
TCFModel model = node.getModel();
- if (ISteppingModeTarget.class == adapterType) {
+ if (IDisassemblyBackend.class == adapterType) {
+ TCFDisassemblyBackend backend = new TCFDisassemblyBackend();
+ if (backend.supportsDebugContext((TCFNode) adaptableObject)) {
+ return backend;
+ }
+ } else if (ISteppingModeTarget.class == adapterType) {
ISteppingModeTarget target = (ISteppingModeTarget) model.getAdapter(adapterType, node);
if (target == null) {
model.setAdapter(adapterType, target = new TCFSteppingModeTarget(model));
@@ -117,6 +129,8 @@ public class TCFNodeAdapterFactory implements IAdapterFactory { if (node instanceof TCFNodeExpression) {
return new TCFWatchpointTarget((TCFNodeExpression) node);
}
+ } else if (ISourceNotFoundPresentation.class == adapterType) {
+ return fgSourceNotFoundPresentation;
}
}
return null;
diff --git a/plugins/org.eclipse.tm.tcf.cdt.ui/src/org/eclipse/tm/internal/tcf/cdt/ui/sourcelookup/TCFCSourceNotFoundEditor.java b/plugins/org.eclipse.tm.tcf.cdt.ui/src/org/eclipse/tm/internal/tcf/cdt/ui/sourcelookup/TCFCSourceNotFoundEditor.java new file mode 100644 index 000000000..607f7bb52 --- /dev/null +++ b/plugins/org.eclipse.tm.tcf.cdt.ui/src/org/eclipse/tm/internal/tcf/cdt/ui/sourcelookup/TCFCSourceNotFoundEditor.java @@ -0,0 +1,94 @@ +/*******************************************************************************
+ * Copyright (c) 2011 Wind River Systems, Inc. and others.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ * Wind River Systems - initial API and implementation
+ *******************************************************************************/
+package org.eclipse.tm.internal.tcf.cdt.ui.sourcelookup;
+
+import org.eclipse.cdt.debug.core.model.ISteppingModeTarget;
+import org.eclipse.cdt.debug.internal.ui.sourcelookup.CSourceNotFoundEditor;
+import org.eclipse.debug.core.DebugPlugin;
+import org.eclipse.debug.core.ILaunch;
+import org.eclipse.debug.core.ILaunchesListener2;
+import org.eclipse.tm.internal.tcf.cdt.ui.sourcelookup.TCFSourceNotFoundPresentation.TCFCSourceNotFoundEditorInput;
+import org.eclipse.tm.internal.tcf.cdt.ui.sourcelookup.TCFSourceNotFoundPresentation.TCFCSourceNotFoundElement;
+import org.eclipse.tm.internal.tcf.debug.ui.model.TCFModel;
+import org.eclipse.ui.IEditorInput;
+import org.eclipse.ui.IEditorPart;
+import org.eclipse.ui.IWorkbenchPage;
+
+/**
+ * Customized source-not-found editor
+ */
+@SuppressWarnings("restriction")
+public class TCFCSourceNotFoundEditor extends CSourceNotFoundEditor {
+
+ static final String ID = "org.eclipse.tm.tcf.cdt.ui.source_not_found_editor";
+
+ private class LaunchListener implements ILaunchesListener2 {
+ public void launchesRemoved(ILaunch[] launches) {
+ }
+ public void launchesAdded(ILaunch[] launches) {
+ }
+ public void launchesChanged(ILaunch[] launches) {
+ }
+ public void launchesTerminated(ILaunch[] launches) {
+ for (ILaunch launch : launches) {
+ if (launch == fLaunch) {
+ closeEditor();
+ break;
+ }
+ }
+ }
+ }
+
+ private ILaunch fLaunch;
+ private LaunchListener fLaunchListener;
+
+ @Override
+ public void setInput(IEditorInput input) {
+ if (input instanceof TCFCSourceNotFoundEditorInput) {
+ TCFCSourceNotFoundElement element = (TCFCSourceNotFoundElement) ((TCFCSourceNotFoundEditorInput) input).getArtifact();
+ TCFModel model = element.getModel();
+ if (model != null) {
+ fLaunch = model.getLaunch();
+ DebugPlugin.getDefault().getLaunchManager().addLaunchListener(fLaunchListener = new LaunchListener());
+ }
+ }
+ super.setInput(input);
+ }
+
+ @Override
+ protected void viewDisassembly() {
+ Object artifact = getArtifact();
+ ISteppingModeTarget target = (ISteppingModeTarget) DebugPlugin.getAdapter(artifact, ISteppingModeTarget.class);
+ if (target != null) {
+ target.enableInstructionStepping(true);
+ closeEditor();
+ }
+ }
+
+ @Override
+ public void dispose() {
+ if (fLaunchListener != null) {
+ DebugPlugin.getDefault().getLaunchManager().removeLaunchListener(fLaunchListener);
+ }
+ super.dispose();
+ }
+
+ protected void closeEditor() {
+ final IEditorPart editor = this;
+ editor.getSite().getShell().getDisplay().asyncExec(new Runnable() {
+ public void run() {
+ IWorkbenchPage page = editor.getSite().getPage();
+ if (page != null) page.closeEditor(editor, false);
+ }
+ });
+ }
+
+}
diff --git a/plugins/org.eclipse.tm.tcf.cdt.ui/src/org/eclipse/tm/internal/tcf/cdt/ui/sourcelookup/TCFSourceNotFoundPresentation.java b/plugins/org.eclipse.tm.tcf.cdt.ui/src/org/eclipse/tm/internal/tcf/cdt/ui/sourcelookup/TCFSourceNotFoundPresentation.java new file mode 100644 index 000000000..11bede84d --- /dev/null +++ b/plugins/org.eclipse.tm.tcf.cdt.ui/src/org/eclipse/tm/internal/tcf/cdt/ui/sourcelookup/TCFSourceNotFoundPresentation.java @@ -0,0 +1,79 @@ +/*******************************************************************************
+ * Copyright (c) 2011 Wind River Systems, Inc. and others.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ * Wind River Systems - initial API and implementation
+ *******************************************************************************/
+package org.eclipse.tm.internal.tcf.cdt.ui.sourcelookup;
+
+import org.eclipse.cdt.debug.internal.core.sourcelookup.CSourceNotFoundElement;
+import org.eclipse.cdt.debug.internal.ui.sourcelookup.CSourceNotFoundEditorInput;
+import org.eclipse.core.runtime.IAdaptable;
+import org.eclipse.debug.core.ILaunchConfiguration;
+import org.eclipse.tm.internal.tcf.debug.ui.model.ISourceNotFoundPresentation;
+import org.eclipse.tm.internal.tcf.debug.ui.model.TCFModel;
+import org.eclipse.ui.IEditorInput;
+
+/**
+ * Reuse CDT's source-not-found editor for TCF.
+ */
+@SuppressWarnings("restriction")
+public class TCFSourceNotFoundPresentation implements ISourceNotFoundPresentation {
+
+ final static class TCFCSourceNotFoundElement extends CSourceNotFoundElement {
+ private final TCFModel fModel;
+
+ private TCFCSourceNotFoundElement(IAdaptable element, ILaunchConfiguration cfg, String file) {
+ super(element, cfg, file);
+ fModel = (TCFModel) element.getAdapter(TCFModel.class);
+ }
+
+ TCFModel getModel() {
+ return fModel;
+ }
+
+ @Override
+ public String getDescription() {
+ return getFile();
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (!(other instanceof TCFCSourceNotFoundElement)) return false;
+ TCFCSourceNotFoundElement otherElement = (TCFCSourceNotFoundElement) other;
+ return getFile().equals(otherElement.getFile()) && getModel() == otherElement.getModel();
+ }
+ }
+
+ final static class TCFCSourceNotFoundEditorInput extends CSourceNotFoundEditorInput {
+ public TCFCSourceNotFoundEditorInput(CSourceNotFoundElement element) {
+ super(element);
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (!(other instanceof TCFCSourceNotFoundEditorInput)) return false;
+ return getArtifact().equals(((TCFCSourceNotFoundEditorInput) other).getArtifact());
+ }
+ }
+
+ public IEditorInput getEditorInput(Object element, ILaunchConfiguration cfg, String file) {
+ if (element instanceof IAdaptable) {
+ return new TCFCSourceNotFoundEditorInput(
+ new TCFCSourceNotFoundElement((IAdaptable) element, cfg, file));
+ }
+ return null;
+ }
+
+ public String getEditorId(IEditorInput input, Object element) {
+ if (input instanceof CSourceNotFoundEditorInput) {
+ return TCFCSourceNotFoundEditor.ID;
+ }
+ return null;
+ }
+
+}
diff --git a/plugins/org.eclipse.tm.tcf.debug.ui/src/org/eclipse/tm/internal/tcf/debug/ui/model/ISourceNotFoundPresentation.java b/plugins/org.eclipse.tm.tcf.debug.ui/src/org/eclipse/tm/internal/tcf/debug/ui/model/ISourceNotFoundPresentation.java new file mode 100644 index 000000000..18c4259b1 --- /dev/null +++ b/plugins/org.eclipse.tm.tcf.debug.ui/src/org/eclipse/tm/internal/tcf/debug/ui/model/ISourceNotFoundPresentation.java @@ -0,0 +1,47 @@ +/*******************************************************************************
+ * Copyright (c) 2011 Wind River Systems, Inc. and others.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ * Wind River Systems - initial API and implementation
+ *******************************************************************************/
+package org.eclipse.tm.internal.tcf.debug.ui.model;
+
+import org.eclipse.debug.core.ILaunchConfiguration;
+import org.eclipse.ui.IEditorInput;
+
+/**
+ * Adapter to customize source presentation for the source-not-found case.
+ */
+public interface ISourceNotFoundPresentation {
+
+ /**
+ * Returns an editor input that should be used to display a source-not-found editor
+ * for the given object or <code>null</code> if unable to provide an editor input
+ * for the given object.
+ *
+ * @param element a debug model element
+ * @param cfg the launch configuration of the debug element
+ * @param file unresolved source file path
+ * @return an editor input, or <code>null</code> if none
+ */
+ public IEditorInput getEditorInput(Object element, ILaunchConfiguration cfg, String file);
+
+ /**
+ * Returns the id of the source-not-found editor to use to display the
+ * given editor input and object, or <code>null</code> if
+ * unable to provide an editor id.
+ *
+ * @param input an editor input that was previously retrieved from this
+ * presentation's <code>getEditorInput</code> method
+ * @param element the object that was used in the call to
+ * <code>getEditorInput</code>, that corresponds to the given editor
+ * input
+ * @return an editor id, or <code>null</code> if none
+ */
+ public String getEditorId(IEditorInput input, Object element);
+
+}
diff --git a/plugins/org.eclipse.tm.tcf.debug.ui/src/org/eclipse/tm/internal/tcf/debug/ui/model/TCFModel.java b/plugins/org.eclipse.tm.tcf.debug.ui/src/org/eclipse/tm/internal/tcf/debug/ui/model/TCFModel.java index 8ad1274c4..6a5886fd9 100644 --- a/plugins/org.eclipse.tm.tcf.debug.ui/src/org/eclipse/tm/internal/tcf/debug/ui/model/TCFModel.java +++ b/plugins/org.eclipse.tm.tcf.debug.ui/src/org/eclipse/tm/internal/tcf/debug/ui/model/TCFModel.java @@ -1057,13 +1057,13 @@ public class TCFModel implements IElementContentProvider, IElementLabelProvider, top_frame = stack_frame.getFrameNo() == 0; ctx_id = stack_frame.parent.id; } - displaySource(cnt, page, ctx_id, top_frame, area); + displaySource(cnt, page, element, ctx_id, top_frame, area); } }); } private void displaySource(final int cnt, final IWorkbenchPage page, - final String exe_id, final boolean top_frame, final ILineNumbers.CodeArea area) { + final Object element, final String exe_id, final boolean top_frame, final ILineNumbers.CodeArea area) { final boolean disassembly_available = channel.getRemoteService(IDisassembly.class) != null; display.asyncExec(new Runnable() { public void run() { @@ -1099,6 +1099,23 @@ public class TCFModel implements IElementContentProvider, IElementLabelProvider, line = area.start_line; } } + if (area != null && !instruction_stepping_enabled && (editor_input == null || editor_id == null)) { + ILaunchConfiguration cfg = launch.getLaunchConfiguration(); + ISourceNotFoundPresentation presentation = (ISourceNotFoundPresentation) DebugPlugin.getAdapter(element, ISourceNotFoundPresentation.class); + if (presentation != null) { + String filename = TCFSourceLookupParticipant.toFileName(area); + editor_input = presentation.getEditorInput(element, cfg, filename); + editor_id = presentation.getEditorId(editor_input, element); + } + if (editor_id == null || editor_input == null) { + editor_id = IDebugUIConstants.ID_COMMON_SOURCE_NOT_FOUND_EDITOR; + editor_input = editor_not_found.get(cfg); + if (editor_input == null) { + editor_input = new CommonSourceNotFoundEditorInput(cfg); + editor_not_found.put(cfg, editor_input); + } + } + } if (exe_id != null && disassembly_available && (editor_input == null || editor_id == null || instruction_stepping_enabled) && PlatformUI.getWorkbench().getEditorRegistry().findEditor( @@ -1106,15 +1123,6 @@ public class TCFModel implements IElementContentProvider, IElementLabelProvider, editor_id = DisassemblyEditorInput.EDITOR_ID; editor_input = DisassemblyEditorInput.INSTANCE; } - if (area != null && (editor_input == null || editor_id == null)) { - ILaunchConfiguration cfg = launch.getLaunchConfiguration(); - editor_id = IDebugUIConstants.ID_COMMON_SOURCE_NOT_FOUND_EDITOR; - editor_input = editor_not_found.get(cfg); - if (editor_input == null) { - editor_input = new CommonSourceNotFoundEditorInput(cfg); - editor_not_found.put(cfg, editor_input); - } - } if (cnt != display_source_generation) return; ITextEditor text_editor = null; if (page != null && editor_input != null && editor_id != null) { diff --git a/python/src/tcf/channel/ChannelProxy.py b/python/src/tcf/channel/ChannelProxy.py new file mode 100644 index 000000000..b95eaeb11 --- /dev/null +++ b/python/src/tcf/channel/ChannelProxy.py @@ -0,0 +1,98 @@ +# *******************************************************************************
+# * Copyright (c) 2011 Wind River Systems, Inc. and others.
+# * All rights reserved. This program and the accompanying materials
+# * are made available under the terms of the Eclipse Public License v1.0
+# * which accompanies this distribution, and is available at
+# * http://www.eclipse.org/legal/epl-v10.html
+# *
+# * Contributors:
+# * Wind River Systems - initial API and implementation
+# *******************************************************************************
+
+"""
+ChannelProxy implements forwarding of TCF messages between two channels.
+The class is used to implement Locator service "redirect" command.
+"""
+
+import exceptions
+from tcf import channel
+
+class ProxyCommandListener(channel.CommandListener):
+ def __init__(self, ch, tokens):
+ self.ch = ch
+ self.tokens = tokens
+ def progress(self, token, data):
+ self.ch.sendProgress(self.tokens.get(token), data)
+ def result(self, token, data):
+ self.ch.sendResult(self.tokens.pop(token, None), data)
+ def terminated(self, token, error):
+ self.ch.rejectCommand(self.tokens.pop(token, None))
+
+class ChannelProxy(object):
+ def __init__(self, x, y):
+ #assert not isinstance(x, ChannelLoop)
+ #assert not isinstance(y, ChannelLoop)
+ self.ch_x = x
+ self.ch_y = y
+ assert self.ch_x.getState() == channel.STATE_OPEN
+ assert self.ch_y.getState() == channel.STATE_OPENING
+ self.tokens_x = {}
+ self.tokens_y = {}
+ cmd_listener_x = ProxyCommandListener(self.ch_x, self.tokens_x)
+ cmd_listener_y = ProxyCommandListener(self.ch_y, self.tokens_y)
+ proxy = self
+
+ class ProxyX(channel.Proxy):
+ def onChannelClosed(self, error):
+ proxy.closed_x = True
+ if proxy.closed_y: return
+ if error is None: proxy.ch_y.close()
+ else: proxy.ch_y.terminate(error)
+ def onCommand(self, token, service, name, data):
+ if proxy.closed_y: return
+ assert proxy.ch_y.getState() == channel.STATE_OPEN
+ s = proxy.ch_y.getRemoteService(service)
+ if not s: proxy.ch_x.terminate(IOError("Invalid service name"))
+ else: proxy.tokens_x[proxy.ch_y.sendCommand(s, name, data, cmd_listener_x)] = token
+ def onEvent(self, service, name, data):
+ s = proxy.ch_x.getRemoteService(service)
+ if not s: proxy.ch_x.terminate(IOError("Invalid service name"))
+ elif not proxy.closed_y: proxy.ch_y.sendEvent(s, name, data)
+
+ class ProxyY(channel.Proxy):
+ def onChannelClosed(self, error):
+ proxy.closed_y = True
+ if proxy.closed_x: return
+ if error is None: proxy.ch_x.close()
+ else: proxy.ch_x.terminate(error)
+ def onCommand(self, token, service, name, data):
+ if proxy.closed_x: return
+ assert proxy.ch_x.getState() == channel.STATE_OPEN
+ s = proxy.ch_x.getRemoteService(service)
+ if not s: proxy.ch_y.terminate(IOError("Invalid service name"))
+ else: proxy.tokens_y[proxy.ch_x.sendCommand(s, name, data, cmd_listener_y)] = token
+ def onEvent(self, service, name, data):
+ s = proxy.ch_y.getRemoteService(service)
+ if not s: proxy.ch_y.terminate(IOError("Invalid service name"))
+ elif not proxy.closed_x: proxy.ch_x.sendEvent(s, name, data)
+
+ proxy_x = ProxyX()
+ proxy_y = ProxyY()
+
+ try:
+ self.ch_y.setProxy(proxy_y, self.ch_x.getRemoteServices())
+ class ChannelListener(channel.ChannelListener):
+ def onChannelClosed(self, error):
+ proxy.ch_y.removeChannelListener(self)
+ if error is None: error = exceptions.Exception("Channel closed")
+ def onChannelOpened(self):
+ proxy.ch_y.removeChannelListener(self)
+ try:
+ proxy.ch_x.setProxy(proxy_x, proxy.ch_y.getRemoteServices())
+ except IOError as e:
+ proxy.ch_x.terminate(e)
+ proxy.ch_y.terminate(e)
+ self.ch_y.addChannelListener(ChannelListener())
+ except IOError as e:
+ self.ch_x.terminate(e)
+ self.ch_y.terminate(e)
diff --git a/python/src/tcf/channel/__init__.py b/python/src/tcf/channel/__init__.py index 0228e5150..dc040aa98 100644 --- a/python/src/tcf/channel/__init__.py +++ b/python/src/tcf/channel/__init__.py @@ -113,6 +113,35 @@ class CommandServer(object): """
pass
+class CommandListener(object):
+ """
+ Command listener interface. Clients implement this interface to
+ receive command results.
+ """
+ def progress(self, token, data):
+ """
+ Called when progress message (intermediate result) is received
+ from remote peer.
+ @param token - command handle
+ @param data - progress message arguments encoded into array of bytes
+ """
+ pass
+ def result(self, token, data):
+ """
+ Called when command result received from remote peer.
+ @param token - command handle
+ @param data - command result message arguments encoded into array of bytes
+ """
+ pass
+ def terminated(self, token, error):
+ """
+ Called when command is terminated because communication channel was closed or
+ command is not recognized by remote peer.
+ @param token - command handle
+ @param error - exception that forced the channel to close
+ """
+ pass
+
def toJSONSequence(args):
if args is None:
return None
diff --git a/python/src/tcf/services/local/LocatorService.py b/python/src/tcf/services/local/LocatorService.py new file mode 100644 index 000000000..3093f3793 --- /dev/null +++ b/python/src/tcf/services/local/LocatorService.py @@ -0,0 +1,842 @@ +# *******************************************************************************
+# * Copyright (c) 2011 Wind River Systems, Inc. and others.
+# * All rights reserved. This program and the accompanying materials
+# * are made available under the terms of the Eclipse Public License v1.0
+# * which accompanies this distribution, and is available at
+# * http://www.eclipse.org/legal/epl-v10.html
+# *
+# * Contributors:
+# * Wind River Systems - initial API and implementation
+# *******************************************************************************
+
+"""
+Locator service uses transport layer to search
+for peers and to collect and maintain up-to-date
+data about peer’s attributes.
+"""
+
+import threading, time, exceptions, socket, cStringIO
+from tcf.services import locator
+from tcf.util import logging
+from tcf.channel import fromJSONSequence, toJSONSequence
+from tcf.channel.ChannelProxy import ChannelProxy
+from tcf import protocol, peer, errors
+
+# Flag indicating whether tracing of the the discovery activity is enabled.
+__TRACE_DISCOVERY = False
+
+class SubNet(object):
+ def __init__(self, prefix_length, address, broadcast):
+ self.prefix_length = prefix_length
+ self.address = address
+ self.broadcast = broadcast
+
+ def contains(self, addr):
+ if addr is None or self.address is None: return False
+ a1 = addr.getAddress()
+ a2 = self.address.getAddress()
+ if len(a1) != len(a2): return False
+ i = 0
+ if self.prefix_length <= len(a1) * 8:
+ l = self.prefix_length
+ else:
+ l = len(a1) * 8
+ while i + 8 <= l:
+ n = i / 8
+ if a1[n] != a2[n]: return False
+ i += 8
+ while i < l:
+ n = i / 8
+ m = 1 << (7 - i % 8)
+ if (a1[n] & m) != (a2[n] & m): return False
+ i += 1
+ return True
+
+ def __eq__(self, o):
+ if not isinstance(o, SubNet): return False
+ return self.prefix_length == o.prefix_length and \
+ self.broadcast == o.broadcast and \
+ self.address == o.address
+
+ def __hash__(self):
+ return hash(self.address)
+
+ def __str__(self):
+ return "%s/%d" % (self.address.getHostAddress(), self.prefix_length)
+
+class Slave(object):
+ # Time of last packet receiver from self slave
+ last_packet_time = 0
+
+ #Time of last REQ_SLAVES packet received from self slave
+ last_req_slaves_time = 0
+
+ def __init__(self, address, port):
+ self.address = address
+ self.port = port
+
+ def __str__(self):
+ return "%s/%d" % (self.address.getHostAddress(), self.port)
+
+class AddressCacheItem(object):
+ address = None
+ time_stamp = 0
+ used = False
+ def __init__(self, host):
+ self.host = host
+
+class InetAddress(object):
+ "Mimicking Java InetAddress class"
+ def __init__(self, host, addr):
+ self.host = host
+ self.addr = addr
+ def getHostAddress(self):
+ return self.addr
+
+class InputPacket(object):
+ """
+ Wrapper for final class DatagramPacket so its toString() can present
+ the value in the debugger in a readable fashion.
+ """
+ def __init__(self, data, addr):
+ self.data = data
+ self.addr = addr
+
+ def getLength(self):
+ return len(self.data)
+
+ def getData(self):
+ return self.data
+
+ def getPort(self):
+ return self.addr[1]
+
+ def getAddress(self):
+ return self.addr[0]
+
+ def __str__(self):
+ return "[address=%s,port=%d,data=\"%s\"]" % (self.getAddress(), self.getPort(), self.data)
+
+DISCOVEY_PORT = 1534
+MAX_PACKET_SIZE = 9000 - 40 - 8
+PREF_PACKET_SIZE = 1500 - 40 - 8
+
+# TODO: research usage of DNS-SD (DNS Service Discovery) to discover TCF peers
+class LocatorService(locator.LocatorService):
+ locator = None
+ peers = {} # str->Peer
+ listeners = [] # list of LocatorListener
+ error_log = set() # set of str
+ _error_log_lock = threading.RLock()
+
+ addr_cache = {} # str->AddressCacheItem
+ _addr_cache_lock = threading.Condition()
+ addr_request = False
+ local_peer = None
+ last_master_packet_time = 0
+
+ def __init__(self):
+ self.subnets = set()
+ self.slaves = []
+ self.inp_buf = bytearray(MAX_PACKET_SIZE)
+ self.out_buf = bytearray(MAX_PACKET_SIZE)
+ service = self
+ LocatorService.locator = self
+ LocatorService.local_peer = peer.LocalPeer()
+ class TimerThread(threading.Thread):
+ def __init__(self, callable):
+ self._callable = callable
+ def __call__(self):
+ while True:
+ try:
+ time.sleep(locator.DATA_RETENTION_PERIOD / 4 / 1000.)
+ protocol.invokeAndWait(self._callable)
+ except RuntimeError:
+ # TCF event dispatch is shut down
+ return
+ except exceptions.Exception as x:
+ service.log("Unhandled exception in TCF discovery timer thread", x)
+ self.timer_thread = TimerThread(self.__refresh_timer)
+ class DNSLookupThread(threading.Thread):
+ def __call__(self):
+ while True:
+ try:
+ itemSet = None
+ with LocatorService._addr_cache_lock:
+ if not LocatorService.addr_request:
+ LocatorService._addr_cache_lock.wait(locator.DATA_RETENTION_PERIOD)
+ msec = int(time.time())
+ for a in LocatorService.addr_cache.values():
+ if a.time_stamp + locator.DATA_RETENTION_PERIOD * 10 < msec:
+ if a.used:
+ if itemSet is None: itemSet = set()
+ itemSet.add(a)
+ else:
+ LocatorService.addr_cache.remove(a)
+ LocatorService.addr_request = False
+ if itemSet is not None:
+ for a in itemSet:
+ addr = None
+ try:
+ addr = socket.gethostbyname(a.host)
+ except socket.gaierror:
+ pass
+ with LocatorService._addr_cache_lock:
+ if addr is None:
+ a.address = None
+ else:
+ a.address = InetAddress(a.host, addr)
+ a.time_stamp = time
+ a.used = False
+ except exceptions.BaseException as x:
+ service.log("Unhandled exception in TCF discovery DNS lookup thread", x)
+ self.dns_lookup_thread = DNSLookupThread()
+ class InputThread(threading.Thread):
+ def __call__(self):
+ try:
+ while True:
+ sock = service.socket
+ try:
+ data, addr = sock.recvfrom(MAX_PACKET_SIZE)
+ p = InputPacket(data, addr)
+ protocol.invokeAndWait(self._callable, p)
+ except RuntimeError:
+ # TCF event dispatch is shutdown
+ return
+ except exceptions.Exception as x:
+ if sock != service.socket: continue
+ port = sock.getsockname()[1]
+ service.log("Cannot read from datagram socket at port %d" % port, x)
+ time.sleep(2)
+ except exceptions.BaseException as x:
+ service.log("Unhandled exception in socket reading thread", x)
+ self.input_thread = InputThread()
+ try:
+ self.loopback_addr = socket.gethostname()
+ self.out_buf[0:8] = 'TCF%s\0\0\0\0' % locator.CONF_VERSION
+ self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+ try:
+ self.socket.bind(('', DISCOVEY_PORT))
+ if __TRACE_DISCOVERY:
+ logging.trace("Became the master agent (bound to port %d)" % self.socket.getsockname()[1])
+ except socket.error as x:
+ self.socket.bind(('', 0))
+ if __TRACE_DISCOVERY:
+ logging.trace("Became a slave agent (bound to port %d)" + self.socket.getsockname()[1])
+ self.socket.setsockopt(socket.SOL_UDP, socket.SO_BROADCAST, 1)
+ self.input_thread.setName("TCF Locator Receiver")
+ self.timer_thread.setName("TCF Locator Timer")
+ self.dns_lookup_thread.setName("TCF Locator DNS Lookup")
+ self.input_thread.setDaemon(True)
+ self.timer_thread.setDaemon(True)
+ self.dns_lookup_thread.setDaemon(True)
+ self.input_thread.start()
+ self.timer_thread.start()
+ self.dns_lookup_thread.start()
+ class LocatorListener(locator.LocatorListener):
+ def peerAdded(self, peer):
+ service._sendPeerInfo(peer, None, 0)
+ def peerChanged(self, peer):
+ service._sendPeerInfo(peer, None, 0)
+ self.listeners.add(LocatorListener())
+ self.__refreshSubNetList()
+ self.__sendPeersRequest(None, 0)
+ self.__sendAll(None, 0, None, int(time.time()))
+ except exceptions.Exception as x:
+ self.log("Cannot open UDP socket for TCF discovery protocol", x)
+
+ @classmethod
+ def getLocalPeer(cls):
+ return cls.local_peer
+
+ @classmethod
+ def getListeners(cls):
+ return cls.listeners[:]
+
+ def __makeErrorReport(self, code, msg):
+ err = {}
+ err[errors.ERROR_TIME] = int(time.time())
+ err[errors.ERROR_CODE] = code
+ err[errors.ERROR_FORMAT] = msg
+ return err
+
+ def __command(self, channel, token, name, data):
+ try:
+ if name == "redirect":
+ peer_id = fromJSONSequence(data)[0]
+ _peer = self.peers.get(peer_id)
+ if _peer is None:
+ error = self.__makeErrorReport(errors.TCF_ERROR_UNKNOWN_PEER, "Unknown peer ID")
+ channel.sendResult(token, toJSONSequence((error,)))
+ return
+ channel.sendResult(token, toJSONSequence((None,)))
+ if isinstance(_peer, peer.LocalPeer):
+ channel.sendEvent(protocol.getLocator(), "Hello", toJSONSequence((channel.getLocalServices(),)))
+ return
+ ChannelProxy(channel, _peer.openChannel())
+ elif name == "sync":
+ channel.sendResult(token, None)
+ elif name == "getPeers":
+ arr = []
+ for p in self.peers.values():
+ arr.append(p.getAttributes())
+ channel.sendResult(token, toJSONSequence((None, arr)))
+ else:
+ channel.rejectCommand(token)
+ except exceptions.BaseException as x:
+ channel.terminate(x)
+
+ def _log(self, msg, x):
+ # Don't report same error multiple times to avoid filling up the log file.
+ with self._error_log_lock:
+ if msg in self.error_log: return
+ self.error_log.add(msg)
+ protocol.log(msg, x)
+
+ def _getInetAddress(self, host):
+ if not host: return None
+ with self._addr_cache_lock:
+ i = self.addr_cache.get(host)
+ if i is None:
+ i = AddressCacheItem(host)
+ ch = host[0]
+ if ch == '[' or ch == ':' or ch >= '0' and ch <= '9':
+ try:
+ addr = socket.gethostbyname(host)
+ i.address = InetAddress(host, addr)
+ except socket.gaierror:
+ pass
+ i.time_stamp = int(time.time())
+ else:
+ # socket.gethostbyname() can cause long delay - delegate to background thread
+ LocatorService.addr_request = True
+ self._addr_cache_lock.notify()
+ self.addr_cache.put(host, i)
+ i.used = True
+ return i.address
+
+ def __refresh_timer(self):
+ tm = int(time.time())
+ # Cleanup slave table
+ if self.slaves:
+ i = 0
+ while i < len(self.slaves):
+ s = self.slaves[i]
+ if s.last_packet_time + locator.DATA_RETENTION_PERIOD < tm:
+ del self.slaves
+ else:
+ i += 1
+
+ # Cleanup peers table
+ stale_peers = None
+ for p in self.peers.values():
+ if isinstance(p, peer.RemotePeer):
+ if p.getLastUpdateTime() + locator.DATA_RETENTION_PERIOD < time:
+ if stale_peers == None: stale_peers = []
+ stale_peers.append(p)
+ if stale_peers is not None:
+ for p in stale_peers: p.dispose()
+
+ # Try to become a master
+ port = self.socket.getsockname()[1]
+ if port != DISCOVEY_PORT and \
+ self.last_master_packet_time + locator.DATA_RETENTION_PERIOD / 2 <= tm:
+ s0 = self.socket
+ s1 = None
+ try:
+ s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+ s1.bind(DISCOVEY_PORT)
+ s1.setsockopt(socket.SOL_UDP, socket.SO_BROADCAST, 1)
+ self.socket = s1
+ s0.close()
+ except:
+ pass
+ self.__refreshSubNetList()
+ if port != DISCOVEY_PORT:
+ for subnet in self.subnets:
+ self.__addSlave(subnet.address, port, tm, tm)
+ self.__sendAll(None, 0, None, tm)
+
+ def __addSlave(self, addr, port, timestamp, time_now):
+ for s in self.slaves:
+ if s.port == port and s.address == addr:
+ if s.last_packet_time < timestamp: s.last_packet_time = timestamp
+ return s
+ s = Slave(addr, port)
+ s.last_packet_time = timestamp
+ self.slaves.append(s)
+ self.__sendPeersRequest(addr, port)
+ self.__sendAll(addr, port, s, time_now)
+ self.__sendSlaveInfo(s, time_now)
+ return s
+
+ def __refreshSubNetList(self):
+ subNetSet = set()
+ try:
+ self.__getSubNetList(subNetSet)
+ except exceptions.BaseException as x:
+ self.log("Cannot get list of network interfaces", x)
+ for s in self.subnets:
+ if s in subNetSet: continue
+ self.subnets.remove(s)
+ for s in subNetSet:
+ if s in self.subnets: continue
+ self.subnets.append(s)
+ if __TRACE_DISCOVERY:
+ str = cStringIO.StringIO()
+ str.write("Refreshed subnet list:")
+ for subnet in self.subnets:
+ str.write("\n\t* address=%s, broadcast=%s" + (subnet.address, subnet.broadcast))
+ logging.trace(str.getvalue())
+
+ def __getSubNetList(self, set):
+ name, aliases, addresses = socket.gethostbyname_ex(socket.gethostname())
+ for address in addresses:
+ rawaddr = socket.inet_aton(address)
+ if len(rawaddr) != 4: continue
+ rawaddr = rawaddr[:3] + '\0xFF'
+ broadcast = socket.inet_ntoa(rawaddr)
+ set.append(SubNet(24, address, broadcast))
+
+ def __getUTF8Bytes(self, s):
+ return s.encode("UTF-8")
+
+ # Used for tracing
+ packetTypes = [
+ None,
+ "CONF_REQ_INFO",
+ "CONF_PEER_INFO",
+ "CONF_REQ_SLAVES",
+ "CONF_SLAVES_INFO"
+ ]
+
+ def __sendDatagramPacket(self, subnet, size, addr, port):
+ try:
+ if addr is None:
+ addr = subnet.broadcast
+ port = DISCOVEY_PORT
+ for slave in self.slaves:
+ self.__sendDatagramPacket(subnet, size, slave.address, slave.port)
+ if not subnet.contains(addr): return False
+ if port == self.socket.getsockname()[1] and addr == subnet.address: return False
+ self.socket.send(str(self.out_buf[:size]), (addr, port))
+
+ if __TRACE_DISCOVERY:
+ map = None
+ if self.out_buf[4] == locator.CONF_PEER_INFO:
+ map = self.__parsePeerAttributes(self.out_buf, 8)
+ self.__traceDiscoveryPacket(False, self.packetTypes[self.out_buf[4]], map, addr, port)
+ except exceptions.BaseException as x:
+ self.log("Cannot send datagram packet to %s" % addr, x)
+ return False
+ return True
+
+ def __parsePeerAttributes(self, data, size):
+ """
+ Parse peer attributes in CONF_INFO_PEER packet data
+
+ @param data
+ the packet section that contain the peer attributes
+ @param size
+ the number of bytes in [data] that contain peer attributes
+ @return a map containing the attributes
+ """
+ map = {}
+ s = data[8:size - 8].decode("UTF-8")
+ l = len(s)
+ i = 0
+ while i < l:
+ i0 = i
+ while i < l and s[i] != '=' and s[i] != '\0': i += 1
+ i1 = i
+ if i < l and s[i] == '=': i += 1
+ i2 = i
+ while i < l and s[i] != '\0': i += 1
+ i3 = i
+ if i < l and s[i] == '\0': i += 1
+ key = s[i0:i1]
+ val = s[i2:i3]
+ map.put(key, val)
+ return map
+
+ def __sendPeersRequest(self, addr, port):
+ self.out_buf[4] = locator.CONF_REQ_INFO
+ for subnet in self.subnets:
+ self.__sendDatagramPacket(subnet, 8, addr, port)
+
+ def __sendPeerInfo(self, _peer, addr, port):
+ attrs = _peer.getAttributes()
+ peer_addr = self.__getInetAddress(attrs.get(peer.ATTR_IP_HOST))
+ if peer_addr is None: return
+ if attrs.get(peer.ATTR_IP_PORT) is None: return
+ self.out_buf[4] = CONF_PEER_INFO
+ i = 8
+
+ for subnet in self.subnets:
+ if isinstance(_peer, peer.RemotePeer):
+ if self.socket.getsockname()[1] != DISCOVEY_PORT: return
+ if (!subnet.address.equals(loopback_addr) && !subnet.address.equals(peer_addr)) continue
+ }
+ if (!subnet.address.equals(loopback_addr)) {
+ if (!subnet.contains(peer_addr)) continue
+ }
+ if (i == 8) {
+ StringBuffer sb = new StringBuffer(out_buf.length)
+ for (str key : attrs.keySet()) {
+ sb.append(key)
+ sb.append('=')
+ sb.append(attrs.get(key))
+ sb.append((char)0)
+ }
+ byte[] bt = getUTF8Bytes(sb.toString())
+ if (i + bt.length > out_buf.length) return
+ System.arraycopy(bt, 0, out_buf, i, bt.length)
+ i += bt.length
+ }
+ if (sendDatagramPacket(subnet, i, addr, port)) subnet.send_all_ok = True
+ }
+ }
+
+ private void sendEmptyPacket(InetAddress addr, int port) {
+ out_buf[4] = CONF_SLAVES_INFO
+ for (SubNet subnet : subnets) {
+ if (subnet.send_all_ok) continue
+ sendDatagramPacket(subnet, 8, addr, port)
+ }
+ }
+
+ private void sendAll(InetAddress addr, int port, Slave sl, long time) {
+ for (SubNet subnet : subnets) subnet.send_all_ok = False
+ for (IPeer peer : peers.values()) sendPeerInfo(peer, addr, port)
+ if (addr != None && sl != None && sl.last_req_slaves_time + DATA_RETENTION_PERIOD >= time) {
+ sendSlavesInfo(addr, port, time)
+ }
+ sendEmptyPacket(addr, port)
+ }
+
+ private void sendSlavesRequest(SubNet subnet, InetAddress addr, int port) {
+ out_buf[4] = CONF_REQ_SLAVES
+ sendDatagramPacket(subnet, 8, addr, port)
+ }
+
+ private void sendSlaveInfo(Slave x, long time) {
+ int ttl = (int)(x.last_packet_time + DATA_RETENTION_PERIOD - time)
+ if (ttl <= 0) return
+ out_buf[4] = CONF_SLAVES_INFO
+ for (SubNet subnet : subnets) {
+ if (!subnet.contains(x.address)) continue
+ int i = 8
+ str s = ttl + ":" + x.port + ":" + x.address.getHostAddress()
+ byte[] bt = getUTF8Bytes(s)
+ System.arraycopy(bt, 0, out_buf, i, bt.length)
+ i += bt.length
+ out_buf[i++] = 0
+ for (Slave y : slaves) {
+ if (!subnet.contains(y.address)) continue
+ if (y.last_req_slaves_time + DATA_RETENTION_PERIOD < time) continue
+ sendDatagramPacket(subnet, i, y.address, y.port)
+ }
+ }
+ }
+
+ private void sendSlavesInfo(InetAddress addr, int port, long time) {
+ out_buf[4] = CONF_SLAVES_INFO
+ for (SubNet subnet : subnets) {
+ if (!subnet.contains(addr)) continue
+ int i = 8
+ for (Slave x : slaves) {
+ int ttl = (int)(x.last_packet_time + DATA_RETENTION_PERIOD - time)
+ if (ttl <= 0) continue
+ if (x.port == port && x.address.equals(addr)) continue
+ if (!subnet.address.equals(loopback_addr)) {
+ if (!subnet.contains(x.address)) continue
+ }
+ subnet.send_all_ok = True
+ str s = x.last_packet_time + ":" + x.port + ":" + x.address.getHostAddress()
+ byte[] bt = getUTF8Bytes(s)
+ if (i > 8 && i + bt.length >= PREF_PACKET_SIZE) {
+ sendDatagramPacket(subnet, i, addr, port)
+ i = 8
+ }
+ System.arraycopy(bt, 0, out_buf, i, bt.length)
+ i += bt.length
+ out_buf[i++] = 0
+ }
+ if (i > 8) sendDatagramPacket(subnet, i, addr, port)
+ }
+ }
+
+ private boolean isRemote(InetAddress address, int port) {
+ if (port != socket.getLocalPort()) return True
+ for (SubNet s : subnets) {
+ if (s.address.equals(address)) return False
+ }
+ return True
+ }
+
+ private void handleDatagramPacket(InputPacket p) {
+ try {
+ long time = System.currentTimeMillis()
+ byte[] buf = p.getData()
+ int len = p.getLength()
+ if (len < 8) return
+ if (buf[0] != 'T') return
+ if (buf[1] != 'C') return
+ if (buf[2] != 'F') return
+ if (buf[3] != CONF_VERSION) return
+ int remote_port = p.getPort()
+ InetAddress remote_address = p.getAddress()
+ if (isRemote(remote_address, remote_port)) {
+ Slave sl = None
+ if (remote_port != DISCOVEY_PORT) {
+ sl = addSlave(remote_address, remote_port, time, time)
+ }
+ switch (buf[4]) {
+ case CONF_PEER_INFO:
+ handlePeerInfoPacket(p)
+ break
+ case CONF_REQ_INFO:
+ handleReqInfoPacket(p, sl, time)
+ break
+ case CONF_SLAVES_INFO:
+ handleSlavesInfoPacket(p, time)
+ break
+ case CONF_REQ_SLAVES:
+ handleReqSlavesPacket(p, sl, time)
+ break
+ }
+ for (SubNet subnet : subnets) {
+ if (!subnet.contains(remote_address)) continue
+ long delay = DATA_RETENTION_PERIOD / 3
+ if (remote_port != DISCOVEY_PORT) delay = DATA_RETENTION_PERIOD / 32
+ else if (!subnet.address.equals(remote_address)) delay = DATA_RETENTION_PERIOD / 2
+ if (subnet.last_slaves_req_time + delay <= time) {
+ sendSlavesRequest(subnet, remote_address, remote_port)
+ subnet.last_slaves_req_time = time
+ }
+ if (subnet.address.equals(remote_address) && remote_port == DISCOVEY_PORT) {
+ last_master_packet_time = time
+ }
+ }
+ }
+ }
+ catch (Throwable x) {
+ log("Invalid datagram packet received from " + p.getAddress() + "/" + p.getPort(), x)
+ }
+ }
+
+ private void handlePeerInfoPacket(InputPacket p) {
+ try {
+ Map<str,str> map = parsePeerAtrributes(p.getData(), p.getLength())
+ if (TRACE_DISCOVERY) traceDiscoveryPacket(True, "CONF_PEER_INFO", map, p)
+ str id = map.get(IPeer.ATTR_ID)
+ if (id == None) throw new Exception("Invalid peer info: no ID")
+ boolean ok = True
+ str host = map.get(IPeer.ATTR_IP_HOST)
+ if (host != None) {
+ ok = False
+ InetAddress peer_addr = getInetAddress(host)
+ if (peer_addr != None) {
+ for (SubNet subnet : subnets) {
+ if (subnet.contains(peer_addr)) {
+ ok = True
+ break
+ }
+ }
+ }
+ }
+ if (ok) {
+ IPeer peer = peers.get(id)
+ if (peer instanceof RemotePeer) {
+ ((RemotePeer)peer).updateAttributes(map)
+ }
+ else if (peer == None) {
+ new RemotePeer(map)
+ }
+ }
+ }
+ catch (Exception x) {
+ log("Invalid datagram packet received from " + p.getAddress() + "/" + p.getPort(), x)
+ }
+ }
+
+ private void handleReqInfoPacket(InputPacket p, Slave sl, long time) {
+ if (TRACE_DISCOVERY) {
+ traceDiscoveryPacket(True, "CONF_REQ_INFO", None, p)
+ }
+ sendAll(p.getAddress(), p.getPort(), sl, time)
+ }
+
+ private void handleSlavesInfoPacket(InputPacket p, long time_now) {
+ try {
+ Map<str,str> trace_map = None # used for tracing only
+ int slave_index = 0 # used for tracing only
+ if (TRACE_DISCOVERY) {
+ trace_map = new HashMap<str,str>(3)
+ }
+
+ str s = new str(p.getData(), 8, p.getLength() - 8, "UTF-8")
+ int l = s.length()
+ int i = 0
+ while (i < l) {
+ int time0 = i
+ while (i < l&& s.charAt(i) != ':' && s.charAt(i) != 0) i++
+ int time1 = i
+ if (i < l && s.charAt(i) == ':') i++
+ int port0 = i
+ while (i < l&& s.charAt(i) != ':' && s.charAt(i) != 0) i++
+ int port1 = i
+ if (i < l && s.charAt(i) == ':') i++
+ int host0 = i
+ while (i < l && s.charAt(i) != 0) i++
+ int host1 = i
+ if (i < l && s.charAt(i) == 0) i++
+ int port = Integer.parseInt(s.substring(port0, port1))
+ str timestamp = s.substring(time0, time1)
+ str host = s.substring(host0, host1)
+ if (TRACE_DISCOVERY) {
+ trace_map.put("slave[" + slave_index++ + ']', timestamp + ':' + port + ':' + host)
+ }
+ if (port != DISCOVEY_PORT) {
+ InetAddress addr = getInetAddress(host)
+ if (addr != None) {
+ long delta = 10006030 # 30 minutes
+ long time_val = timestamp.length() > 0 ? Long.parseLong(timestamp) : time_now
+ if (time_val < 3600000) {
+ """Time stamp is "time to live" in milliseconds"""
+ time_val = time_now + time_val / 1000 - DATA_RETENTION_PERIOD
+ }
+ else if (time_val < time_now / 1000 + 50000000) {
+ """Time stamp is in seconds"""
+ time_val= 1000
+ }
+ else {
+ """Time stamp is in milliseconds"""
+ }
+ if (time_val < time_now - delta || time_val > time_now + delta) {
+ SimpleDateFormat fmt = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")
+ str msg =
+ "Invalid slave info timestamp: " + timestamp +
+ " -> " + fmt.format(new Date(time_val))
+ log("Invalid datagram packet received from " +
+ p.getAddress() + "/" + p.getPort(),
+ new Exception(msg))
+ time_val = time_now - DATA_RETENTION_PERIOD / 2
+ }
+ addSlave(addr, port, time_val, time_now)
+ }
+ }
+ }
+ if (TRACE_DISCOVERY) {
+ traceDiscoveryPacket(True, "CONF_SLAVES_INFO", trace_map, p)
+ }
+ }
+ catch (Exception x) {
+ log("Invalid datagram packet received from " + p.getAddress() + "/" + p.getPort(), x)
+ }
+ }
+
+ private void handleReqSlavesPacket(InputPacket p, Slave sl, long time) {
+ if (TRACE_DISCOVERY) {
+ traceDiscoveryPacket(True, "CONF_REQ_SLAVES", None, p)
+ }
+ if (sl != None) sl.last_req_slaves_time = time
+ sendSlavesInfo(p.getAddress(), p.getPort(), time)
+ }
+
+ """----------------------------------------------------------------------------------"""
+
+ public static LocatorService getLocator() {
+ return locator
+ }
+
+ public str getName() {
+ return NAME
+ }
+
+ public Map<str,IPeer> getPeers() {
+ assert Protocol.isDispatchThread()
+ return peers
+ }
+
+ public IToken redirect(str peer_id, DoneRedirect done) {
+ throw new Error("Channel redirect cannot be done on local peer")
+ }
+
+ public IToken redirect(Map<str,str> peer, DoneRedirect done) {
+ throw new Error("Channel redirect cannot be done on local peer")
+ }
+
+ public IToken sync(DoneSync done) {
+ throw new Error("Channel sync cannot be done on local peer")
+ }
+
+ public void addListener(LocatorListener listener) {
+ assert listener != None
+ assert Protocol.isDispatchThread()
+ listeners.add(listener)
+ }
+
+ public void removeListener(LocatorListener listener) {
+ assert Protocol.isDispatchThread()
+ listeners.remove(listener)
+ }
+
+ """
+ Log that a TCF Discovery packet has be sent or received. The trace is
+ sent to stdout. This should be called only if the tracing has been turned
+ on via java property definitions.
+
+ @param received
+ True if the packet was sent, otherwise it was received
+ @param type
+ a string specifying the type of packet, e.g., "CONF_PEER_INFO"
+ @param attrs
+ a set of attributes relevant to the type of packet (typically
+ a peer's attributes)
+ @param addr
+ the network address the packet is being sent to
+ @param port
+ the port the packet is being sent to
+ """
+ private static void traceDiscoveryPacket(boolean received, str type, Map<str,str> attrs, InetAddress addr, int port) {
+ assert TRACE_DISCOVERY
+ StringBuilder str = new StringBuilder(type + (received ? " received from " : " sent to ") + addr + "/" + port)
+ if (attrs != None) {
+ Iterator<Entry<str, str>> iter = attrs.entrySet().iterator()
+ while (iter.hasNext()) {
+ Entry<str, str> entry = iter.next()
+ str.append("\n\t" + entry.getKey() + '=' + entry.getValue())
+ }
+ }
+ logging.trace(str.toString())
+ }
+
+ """
+ Convenience variant that takes a DatagramPacket for specifying
+ the target address and port.
+ """
+ private static void traceDiscoveryPacket(boolean received, str type, Map<str,str> attrs, InputPacket packet) {
+ traceDiscoveryPacket(received, type, attrs, packet.getAddress(), packet.getPort())
+ }
+}
+
+
+static {
+ ServiceManager.addServiceProvider(new IServiceProvider() {
+
+ public IService[] getLocalService(final IChannel channel) {
+ channel.addCommandServer(locator, new IChannel.ICommandServer() {
+ public void command(IToken token, str name, byte[] data) {
+ locator.command((AbstractChannel)channel, token, name, data)
+ }
+ })
+ return new IService[]{ locator }
+ }
+
+ public IService getServiceProxy(IChannel channel, str service_name) {
+ return None
+ }
+ })
+}
+
diff --git a/python/src/tcf/services/local/__init__.py b/python/src/tcf/services/local/__init__.py new file mode 100644 index 000000000..cf36110a7 --- /dev/null +++ b/python/src/tcf/services/local/__init__.py @@ -0,0 +1,10 @@ +# *******************************************************************************
+# * Copyright (c) 2011 Wind River Systems, Inc. and others.
+# * All rights reserved. This program and the accompanying materials
+# * are made available under the terms of the Eclipse Public License v1.0
+# * which accompanies this distribution, and is available at
+# * http://www.eclipse.org/legal/epl-v10.html
+# *
+# * Contributors:
+# * Wind River Systems - initial API and implementation
+# *******************************************************************************
diff --git a/python/src/tcf/util/logging.py b/python/src/tcf/util/logging.py new file mode 100644 index 000000000..c4979ede2 --- /dev/null +++ b/python/src/tcf/util/logging.py @@ -0,0 +1,54 @@ +# *******************************************************************************
+# * Copyright (c) 2011 Wind River Systems, Inc. and others.
+# * All rights reserved. This program and the accompanying materials
+# * are made available under the terms of the Eclipse Public License v1.0
+# * which accompanies this distribution, and is available at
+# * http://www.eclipse.org/legal/epl-v10.html
+# *
+# * Contributors:
+# * Wind River Systems - initial API and implementation
+# *******************************************************************************
+
+"Internal utility methods used for logging/tracing."
+
+from tcf import protocol
+import locale, time, cStringIO
+
+DECIMAL_DELIMITER = locale.localeconv().get('decimal_point', '.')
+
+def getDebugTime():
+ """
+ Returns a relative timestamp in the form "seconds,milliseconds". Each
+ segment is zero-padded to three digits, ensuring a consistent length of
+ seven characters. The timestamp has no absolute meaning. It is merely the
+ elapsed time since January 1, 1970 UT truncated at 999 seconds. Do not
+ use this for production code, especially for mathematically determining
+ the relative time between two events, since the counter will flip to zero
+ roughly every 16 minutes.
+ """
+ traceBuilder = cStringIO.StringIO()
+
+ # Record the time
+ tm = int(time.time())
+ seconds = (tm / 1000) % 1000
+ if seconds < 100: traceBuilder.write('0')
+ if seconds < 10: traceBuilder.write('0')
+ traceBuilder.write(str(seconds))
+ traceBuilder.write(DECIMAL_DELIMITER)
+ millis = time % 1000
+ if millis < 100: traceBuilder.write('0')
+ if millis < 10: traceBuilder.write('0')
+ traceBuilder.write(str(millis))
+ return traceBuilder.getvalue()
+
+def trace(msg):
+ """
+ Trace hooks should use this method to log a message. It prepends the
+ message with a timestamp and sends it to the TCF logger facility. The
+ logger implementation may or may not inject its own timestamp. For
+ tracing, we definitely need one, so we introduce a minimal, relative-time
+ stamp.
+
+ @param msg the trace message
+ """
+ protocol.log('%s msg' % (getDebugTime(), msg))
diff --git a/python/todo.twiki b/python/todo.twiki new file mode 100644 index 000000000..602325a87 --- /dev/null +++ b/python/todo.twiki @@ -0,0 +1,39 @@ +---+ TCF Python Scripting
+
+---++ Features
+ * [done] EventQueue
+ * Transport types
+ * [done] TCP
+ * SSL
+ * PIPE
+ * LOOP
+ * Service proxies:
+ * [done] Breakpoints
+ * [done] Locator
+ * [done] RunControl
+ * [done] StackTrace
+ * [done] Symbols
+ * [done] Registers
+ * [done] Expressions
+ * [done] LineNumbers
+ * [done] Processes
+ * [done] FileSystem
+ * MemoryMap
+ * Memory
+ * PathMap
+ * Diagnostics
+ * Disassembly
+ * Streams
+ * Terminals
+ * SysMonitor
+ * Local services:
+ * [incomplete] ZeroCopy
+ * [started] Locator service (discovery)
+ * Utilities:
+ * [done] ACPM data cache
+ * [done] TCFTask
+ * [done] Event recording
+ * [done] Simplified synchronous command invocation
+ * File object implementation based on FileSystem service
+ * [started] Interactive interpreter
+ * JUnit tests
|