Skip to content

Commit

Permalink
#6560: handling combination of code and magic
Browse files Browse the repository at this point in the history
  • Loading branch information
@jaroslawmalekcodete
jaroslawmalekcodete committed Jan 25, 2018
1 parent 322c151 commit 8e6237a
Show file tree
Hide file tree
Showing 47 changed files with 1,000 additions and 550 deletions.
18 changes: 3 additions & 15 deletions kernel/base/src/main/java/com/twosigma/beakerx/evaluator/EvaluatorManager.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -68,9 +68,8 @@ public synchronized void killAllThreads() {
evaluator.killAllThreads();
}

public synchronized SimpleEvaluationObject executeCode(String code, Message message,
int executionCount, KernelFunctionality.ExecuteCodeCallback executeCodeCallback) {
return execute(code, message, executionCount, executeCodeCallback);
public synchronized SimpleEvaluationObject executeCode(String code, SimpleEvaluationObject seo) {
return execute(code, seo);
}

public synchronized SimpleEvaluationObjectWithTime executeCodeWithTimeMeasurement(String code, Message message,
Expand All @@ -82,10 +81,7 @@ public void exit() {
evaluator.exit();
}

private SimpleEvaluationObject execute(String code, Message message, int executionCount,
KernelFunctionality.ExecuteCodeCallback executeCodeCallback) {
SimpleEvaluationObject seo = createSimpleEvaluationObject(code, message, executionCount,
executeCodeCallback);
private SimpleEvaluationObject execute(String code, SimpleEvaluationObject seo) {
evaluator.evaluate(seo, code);
return seo;
}
Expand All @@ -98,14 +94,6 @@ private SimpleEvaluationObjectWithTime executeWithTimeMeasurement(String code, M
return seowt;
}

private SimpleEvaluationObject createSimpleEvaluationObject(String code, Message message,
int executionCount, KernelFunctionality.ExecuteCodeCallback executeCodeCallback) {
SimpleEvaluationObject seo = new SimpleEvaluationObject(code, executeCodeCallback);
seo.setJupyterMessage(message);
seo.setExecutionCount(executionCount);
seo.addObserver(kernel.getExecutionResultSender());
return seo;
}

private SimpleEvaluationObjectWithTime createSimpleEvaluationObjectWithTime(String code, Message message,
int executionCount, KernelFunctionality.ExecuteCodeCallbackWithTime executeCodeCallbackWithTime) {
Expand Down
123 changes: 75 additions & 48 deletions kernel/base/src/main/java/com/twosigma/beakerx/jvm/object/SimpleEvaluationObject.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ public class SimpleEvaluationObject extends Observable {
private BeakerOutputHandler stderr;
private Queue<ConsoleOutput> consoleOutput = new ConcurrentLinkedQueue<ConsoleOutput>();
private ProgressReporting progressReporting;
private boolean showResult = true;

public SimpleEvaluationObject(String e) {
expression = e;
Expand All @@ -66,6 +67,10 @@ public void executeCodeCallback() {
this.executeCodeCallback.execute(this);
}

public boolean isShowResult() {
return showResult;
}

public synchronized void started() {
setOutputHandler();
this.status = EvaluationStatus.RUNNING;
Expand Down Expand Up @@ -110,10 +115,14 @@ public synchronized Object getPayload() {
}

public void structuredUpdate(String message, int progress) {
if(progressReporting ==null){
if (progressReporting == null) {
progressReporting = new ProgressReporting();
}
progressReporting.structuredUpdate(message,progress);
progressReporting.structuredUpdate(message, progress);
}

public void noResult() {
this.showResult = false;
}

public static enum EvaluationStatus {
Expand All @@ -124,13 +133,13 @@ public class SimpleOutputHandler implements BeakerOutputHandler {

private boolean error;

public SimpleOutputHandler(boolean error){
public SimpleOutputHandler(boolean error) {
this.error = error;
}

@Override
public void write(int b) {
byte [] ba = new byte[1];
byte[] ba = new byte[1];
ba[0] = (byte) b;
consoleOutput.add(new ConsoleOutput(error, new String(ba, StandardCharsets.UTF_8)));
setChanged();
Expand Down Expand Up @@ -208,21 +217,27 @@ public String toString() {

private static final int OUTPUT_QUEUE_SIZE = 20;
private static final int MAX_LINE_LENGTH = 240;
private int outputdataCount= 0;
private String buildingout= "";
private List<Object> outputdata= new ArrayList<Object>();
private String buildingerr= "";
private int outputdataCount = 0;
private String buildingout = "";
private List<Object> outputdata = new ArrayList<Object>();
private String buildingerr = "";

public List<Object> getOutputdata() {
return outputdata;
}

public void appendOutput(String s) {
if (getSize() > OUTPUT_QUEUE_SIZE) {
try { Thread.sleep(500); } catch (InterruptedException e) { }
try {
Thread.sleep(500);
} catch (InterruptedException e) {
}
}
if (getSize() > OUTPUT_QUEUE_SIZE) {
try { Thread.sleep(500); } catch (InterruptedException e) { }
try {
Thread.sleep(500);
} catch (InterruptedException e) {
}
}
doAppendOutput(s);
}
Expand All @@ -239,35 +254,35 @@ private synchronized void doAppendOutput(String s) {
add = buildingout;
buildingout = "";
} else {
add = buildingout.substring(0, buildingout.lastIndexOf('\n')+1);
buildingout = buildingout.substring(buildingout.lastIndexOf('\n')+1);
add = buildingout.substring(0, buildingout.lastIndexOf('\n') + 1);
buildingout = buildingout.substring(buildingout.lastIndexOf('\n') + 1);
}
}
if ( buildingout.length() > MAX_LINE_LENGTH) {
if (buildingout.length() > MAX_LINE_LENGTH) {
add = buildingout;
buildingout = "";
}
if (add != null) {
String [] v = add.split("\n");
String[] v = add.split("\n");
for (String sv : v) {
while (sv.length()>MAX_LINE_LENGTH) {
while (sv.length() > MAX_LINE_LENGTH) {
String t = sv.substring(0, MAX_LINE_LENGTH);
sv = sv.substring(MAX_LINE_LENGTH);
if (outputdata.size() == 0 || !(outputdata.get(outputdata.size()-1) instanceof EvaluationStdOutput)) {
outputdata.add(new EvaluationStdOutput(t+"\n"));
if (outputdata.size() == 0 || !(outputdata.get(outputdata.size() - 1) instanceof EvaluationStdOutput)) {
outputdata.add(new EvaluationStdOutput(t + "\n"));
} else {
EvaluationStdOutput st = (EvaluationStdOutput) outputdata.get(outputdata.size()-1);
st.payload += t+"\n";
EvaluationStdOutput st = (EvaluationStdOutput) outputdata.get(outputdata.size() - 1);
st.payload += t + "\n";
}
outputdataCount ++;
outputdataCount++;
}
if (outputdata.size() == 0 || !(outputdata.get(outputdata.size()-1) instanceof EvaluationStdOutput)) {
outputdata.add(new EvaluationStdOutput(sv+"\n"));
if (outputdata.size() == 0 || !(outputdata.get(outputdata.size() - 1) instanceof EvaluationStdOutput)) {
outputdata.add(new EvaluationStdOutput(sv + "\n"));
} else {
EvaluationStdOutput st = (EvaluationStdOutput) outputdata.get(outputdata.size()-1);
st.payload += sv+"\n";
EvaluationStdOutput st = (EvaluationStdOutput) outputdata.get(outputdata.size() - 1);
st.payload += sv + "\n";
}
outputdataCount ++;
outputdataCount++;
}
setChanged();
notifyObservers();
Expand All @@ -276,10 +291,16 @@ private synchronized void doAppendOutput(String s) {

public void appendError(String s) {
if (getSize() > OUTPUT_QUEUE_SIZE) {
try { Thread.sleep(500); } catch (InterruptedException e) { }
try {
Thread.sleep(500);
} catch (InterruptedException e) {
}
}
if (getSize() > OUTPUT_QUEUE_SIZE) {
try { Thread.sleep(500); } catch (InterruptedException e) { }
try {
Thread.sleep(500);
} catch (InterruptedException e) {
}
}
doAppendError(s);
}
Expand All @@ -292,47 +313,47 @@ private synchronized void doAppendError(String s) {
add = buildingerr;
buildingerr = "";
} else {
add = buildingerr.substring(0, buildingerr.lastIndexOf('\n')+1);
buildingerr = buildingerr.substring(buildingerr.lastIndexOf('\n')+1);
add = buildingerr.substring(0, buildingerr.lastIndexOf('\n') + 1);
buildingerr = buildingerr.substring(buildingerr.lastIndexOf('\n') + 1);
}
}
if ( buildingerr.length() > MAX_LINE_LENGTH) {
if (buildingerr.length() > MAX_LINE_LENGTH) {
add = buildingerr;
buildingerr = "";
}
if (add != null) {
/*
* HACK to remove annoying stderr messages from third party libraries
*/
* HACK to remove annoying stderr messages from third party libraries
*/
if ((add.contains("org.antlr.v4.runtime.misc.NullUsageProcessor") && add.contains("'RELEASE_6'")) ||
(add.contains("JavaSourceCompilerImpl compile"))) {
String [] v = add.split("\n");
String[] v = add.split("\n");
add = "";
for(String s2 : v) {
for (String s2 : v) {
if (!s2.contains("org.antlr.v4.runtime.misc.NullUsageProcessor") && !s2.contains("JavaSourceCompilerImpl compile"))
add += s2 + "\n";
}
}
String [] v = add.split("\n");
String[] v = add.split("\n");
for (String sv : v) {
while (sv.length()>MAX_LINE_LENGTH) {
while (sv.length() > MAX_LINE_LENGTH) {
String t = sv.substring(0, MAX_LINE_LENGTH);
sv = sv.substring(MAX_LINE_LENGTH);
if (outputdata.size() == 0 || !(outputdata.get(outputdata.size()-1) instanceof EvaluationStdError)) {
outputdata.add(new EvaluationStdError(t+"\n"));
if (outputdata.size() == 0 || !(outputdata.get(outputdata.size() - 1) instanceof EvaluationStdError)) {
outputdata.add(new EvaluationStdError(t + "\n"));
} else {
EvaluationStdError st = (EvaluationStdError) outputdata.get(outputdata.size()-1);
st.payload += t+"\n";
EvaluationStdError st = (EvaluationStdError) outputdata.get(outputdata.size() - 1);
st.payload += t + "\n";
}
outputdataCount ++;
outputdataCount++;
}
if (outputdata.size() == 0 || !(outputdata.get(outputdata.size()-1) instanceof EvaluationStdError)) {
outputdata.add(new EvaluationStdError(sv+"\n"));
if (outputdata.size() == 0 || !(outputdata.get(outputdata.size() - 1) instanceof EvaluationStdError)) {
outputdata.add(new EvaluationStdError(sv + "\n"));
} else {
EvaluationStdError st = (EvaluationStdError) outputdata.get(outputdata.size()-1);
st.payload += sv+"\n";
EvaluationStdError st = (EvaluationStdError) outputdata.get(outputdata.size() - 1);
st.payload += sv + "\n";
}
outputdataCount ++;
outputdataCount++;
}
setChanged();
notifyObservers();
Expand All @@ -341,12 +362,18 @@ private synchronized void doAppendError(String s) {

public class EvaluationStdOutput {
public String payload;
public EvaluationStdOutput(String s) { payload = s; }

public EvaluationStdOutput(String s) {
payload = s;
}
}

public class EvaluationStdError {
public String payload;
public EvaluationStdError(String s) { payload = s; }

public EvaluationStdError(String s) {
payload = s;
}
}

}
55 changes: 36 additions & 19 deletions kernel/base/src/main/java/com/twosigma/beakerx/kernel/Code.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,54 +15,44 @@
*/
package com.twosigma.beakerx.kernel;

import com.twosigma.beakerx.kernel.magic.command.MagicCommand;
import com.twosigma.beakerx.kernel.magic.command.outcome.MagicCommandOutcomeItem;
import com.twosigma.beakerx.message.Message;

import static com.google.common.base.Preconditions.checkNotNull;
import static com.twosigma.beakerx.kernel.handler.MagicCommandExecutor.sendRepliesWithStatus;
import static org.apache.commons.lang3.builder.EqualsBuilder.reflectionEquals;
import static org.apache.commons.lang3.builder.HashCodeBuilder.reflectionHashCode;
import static org.apache.commons.lang3.builder.ToStringBuilder.reflectionToString;

import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;

public class Code {

private final String allCode;
private final List<MagicCommand> magicCommands;
private Optional<String> codeBlock;
private final List<MagicCommandOutcomeItem> errors;
private final Message message;
private List<CodeFrame> codeFrames;

private Code(String allCode, Optional<String> codeBlock, List<MagicCommand> magicCommands, List<MagicCommandOutcomeItem> errors, Message message) {
private Code(String allCode, List<CodeFrame> codeFrames, List<MagicCommandOutcomeItem> errors, Message message) {
this.allCode = allCode;
this.magicCommands = checkNotNull(magicCommands);
this.codeFrames = checkNotNull(codeFrames);
this.errors = checkNotNull(errors);
this.message = message;
this.codeBlock = codeBlock;
}

public static Code createCode(String allCode, String codeBlock, List<MagicCommand> magicCommands, List<MagicCommandOutcomeItem> errors, Message message) {
return new Code(allCode, Optional.of(codeBlock), magicCommands, errors, message);
public static Code createCode(String allCode, List<CodeFrame> codeFrames, List<MagicCommandOutcomeItem> errors, Message message) {
return new Code(allCode, codeFrames, errors, message);
}

public static Code createCodeWithoutCodeBlock(String allCode, List<MagicCommand> magicCommands, List<MagicCommandOutcomeItem> errors, Message message) {
return new Code(allCode, Optional.empty(), magicCommands, errors, message);
}

public Optional<String> getCodeBlock() {
return codeBlock;
public List<CodeFrame> getCodeFrames() {
return codeFrames;
}

public String asString() {
return this.allCode;
}

public List<MagicCommand> getMagicCommands() {
return magicCommands;
}

public boolean hasErrors() {
return !errors.isEmpty();
}
Expand All @@ -89,4 +79,31 @@ public String toString() {
public Message getMessage() {
return message;
}

public void execute(KernelFunctionality kernel, int executionCount, KernelFunctionality.ExecuteCodeCallback executeCodeCallback) {
if (hasErrors()) {
sendRepliesWithStatus(getErrors(), kernel, getMessage(), executionCount);
executeCodeCallback.execute(null);
} else {
takeCodeFramesWithoutLast()
.forEach(frame -> {
CompletableFuture<Boolean> result = new CompletableFuture<>();
frame.executeFrame(this, kernel, message, executionCount, seo -> result.complete(true));
try {
result.get();
} catch (Exception e) {
throw new RuntimeException(e);
}
});
takeLastCodeFrame().executeLastFrame(this, kernel, message, executionCount, executeCodeCallback);
}
}

private CodeFrame takeLastCodeFrame() {
return getCodeFrames().get(getCodeFrames().size() - 1);
}

private List<CodeFrame> takeCodeFramesWithoutLast() {
return getCodeFrames().subList(0, getCodeFrames().size() - 1);
}
}
Loading

0 comments on commit 8e6237a

Please sign in to comment.