Miroirs/xwords
1
0
Fork 0
mirror of git://xwords.git.sourceforge.net/gitroot/xwords/xwords synced 2025-01-22 07:28:16 +01:00
Code Issues Projects Releases Packages Wiki Activity

prune only those messages successfully sent

Browse source 
Fix race condition that would have cleared all messages including those
added after a send began. In the process, move to storing outbound
messages individually rather than concatenated in their stream format.
This commit is contained in:
Eric House 2019-02-19 08:31:49 -08:00
parent ff2a06592f
commit 341293dc55
1 changed files with 131 additions and 106 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

237
xwords4/android/app/src/main/java/org/eehouse/android/xw4/BTService.java
View file

@ -21,8 +21,6 @@
package org.eehouse.android.xw4; package org.eehouse.android.xw4;
import android.app.Activity; import android.app.Activity;
import android.app.Notification;
import android.app.PendingIntent;
import android.app.Service; import android.app.Service;
import android.bluetooth.BluetoothAdapter; import android.bluetooth.BluetoothAdapter;
import android.bluetooth.BluetoothClass.Device.Major; import android.bluetooth.BluetoothClass.Device.Major;
@ -33,9 +31,7 @@ import android.content.Context;
import android.content.Intent; import android.content.Intent;
import android.os.Build; import android.os.Build;
import android.os.Handler; import android.os.Handler;
import android.os.SystemClock;
import android.provider.Settings; import android.provider.Settings;
import android.support.v4.app.NotificationCompat;
import android.text.TextUtils; import android.text.TextUtils;
import org.eehouse.android.xw4.MultiService.DictFetchOwner; import org.eehouse.android.xw4.MultiService.DictFetchOwner;
@ -98,8 +94,6 @@ public class BTService extends XWJIService {
private static final int MAX_PACKET_LEN = 4 * 1024; private static final int MAX_PACKET_LEN = 4 * 1024;
private static final int BT_PROTO_ORIG = 0; private static final int BT_PROTO_ORIG = 0;
// private static final int BT_PROTO_JSONS = 1; // using jsons instead of lots of fields
private static final int BT_PROTO_NLI = 2; // using binary/common form of NLI
private static final int BT_PROTO_BATCH = 3; private static final int BT_PROTO_BATCH = 3;
private static final int BT_PROTO = BT_PROTO_BATCH; private static final int BT_PROTO = BT_PROTO_BATCH;
@ -1119,22 +1113,49 @@ public class BTService extends XWJIService {
} }
private static class PacketAccumulator { private static class PacketAccumulator {
private static class MsgElem {
BTCmd mCmd;
int mGameID;
long mStamp;
byte[] mData;
MsgElem( BTCmd cmd, int gameID, OutputPair op )
{
mCmd = cmd;
mGameID = gameID;
mStamp = System.currentTimeMillis();
OutputPair tmpOp = new OutputPair();
try {
tmpOp.dos.writeByte( cmd.ordinal() );
byte[] data = op.bos.toByteArray();
tmpOp.dos.writeShort( data.length );
tmpOp.dos.write( data, 0, data.length );
mData = tmpOp.bos.toByteArray();
} catch (IOException ioe ) {
// With memory-backed IO this should be impossible
Log.e( TAG, "MsgElem.__init(): got ioe!: %s",
ioe.getMessage() );
}
}
int size() { return mData.length; }
}
private String mAddr; private String mAddr;
private String mName; private String mName;
private List<BTCmd> mCmds; private List<MsgElem> mElems;
private List<Integer> mGameIDs;
private OutputPair mOP;
private long mStamp;
private long mLastFailTime; private long mLastFailTime;
private int mFailCount; private int mFailCount;
private int mLength;
// private int mPacketCount;
PacketAccumulator( String addr ) { PacketAccumulator( String addr ) {
mAddr = addr; mAddr = addr;
mName = getName( addr ); mName = getName( addr );
clearAll(); mElems = new ArrayList<>();
// Log.d( TAG, "PacketAccumulator.__init() => %s", this ); mFailCount = 0;
mLength = 0;
} }
String getAddr() { return mAddr; } String getAddr() { return mAddr; }
@ -1146,7 +1167,7 @@ public class BTService extends XWJIService {
// Log.d( TAG, "getNextReadyMS() IN" ); // Log.d( TAG, "getNextReadyMS() IN" );
try ( DbgUtils.DeadlockWatch dw = new DbgUtils.DeadlockWatch( this ) ) { try ( DbgUtils.DeadlockWatch dw = new DbgUtils.DeadlockWatch( this ) ) {
synchronized ( this ) { synchronized ( this ) {
if ( 0 == mCmds.size() ) { // nothing to send if ( 0 == mElems.size() ) { // nothing to send
waitFromNow = Long.MAX_VALUE; waitFromNow = Long.MAX_VALUE;
} else if ( 0 == mFailCount ) { } else if ( 0 == mFailCount ) {
waitFromNow = 0; waitFromNow = 0;
@ -1174,10 +1195,17 @@ public class BTService extends XWJIService {
@Override @Override
public synchronized String toString() public synchronized String toString()
{ {
long age = System.currentTimeMillis() - mStamp; long age = 0;
if ( 0 < mElems.size() ) {
age = System.currentTimeMillis() - mElems.get(0).mStamp;
}
List<BTCmd> cmds = new ArrayList<>();
for ( MsgElem elem : mElems ) {
cmds.add( elem.mCmd );
}
return String.format("{name: %s, addr: %s, age: %dms, failCount: %d, len: %d, cmds: %s}", return String.format("{name: %s, addr: %s, age: %dms, failCount: %d, len: %d, cmds: %s}",
mName, mAddr, age, mFailCount, mOP.length(), mName, mAddr, age, mFailCount, mLength,
TextUtils.join(",", mCmds) ); TextUtils.join(",", cmds) );
} }
int writeAndCheck( BluetoothSocket socket, DataOutputStream dos, int writeAndCheck( BluetoothSocket socket, DataOutputStream dos,
@ -1185,50 +1213,40 @@ public class BTService extends XWJIService {
throws IOException throws IOException
{ {
Log.d( TAG, "%s.writeAndCheck() IN", this ); Log.d( TAG, "%s.writeAndCheck() IN", this );
Assert.assertNotNull( dos );
List<BTCmd> localCmds = null; List<MsgElem> localElems = null;
List<Integer> localGameIDs = null;
try ( DbgUtils.DeadlockWatch dw = new DbgUtils.DeadlockWatch( this ) ) { try ( DbgUtils.DeadlockWatch dw = new DbgUtils.DeadlockWatch( this ) ) {
synchronized ( this ) { synchronized ( this ) {
byte[] data = mOP.bos.toByteArray(); if ( 0 < mLength ) {
if ( 0 < data.length ) {
try { try {
// Format is <proto><len-of-rest><msgCount><msgsData> To // Format is <proto><len-of-rest><msgCount><msgsData> To
// insert the count at the beginning we have to create a // insert the count at the beginning we have to create a
// whole new byte array since there's no random access. // whole new byte array since there's no random access.
int msgCount = mElems.size();
OutputPair tmpOP = new OutputPair(); OutputPair tmpOP = new OutputPair();
tmpOP.dos.writeByte( mCmds.size() ); // count of messages tmpOP.dos.writeByte( msgCount ); // count of messages
Log.d( TAG, "writeAndCheck(): wrote msg count: %d", mCmds.size() ); // Log.d( TAG, "writeAndCheck(): wrote msg count: %d", msgCount );
tmpOP.dos.write( data, 0, data.length );
data = tmpOP.bos.toByteArray(); // replace data
// now write to the socket for ( MsgElem elem : mElems ) {
Assert.assertNotNull( dos ); byte[] elemData = elem.mData;
// dos.writeByte( BT_PROTO ); tmpOP.dos.write( elemData, 0, elemData.length );
}
byte[] data = tmpOP.bos.toByteArray();
// now write to the socket. Note that connect()
// writes BT_PROTO as the first byte.
dos.writeShort( data.length ); dos.writeShort( data.length );
dos.write( data, 0, data.length ); dos.write( data, 0, data.length );
dos.flush(); dos.flush();
Log.d( TAG, "writeAndCheck(): wrote %d-byte payload with sum %s", Log.d( TAG, "writeAndCheck(): wrote %d msgs as"
data.length, Utils.getMD5SumFor( data ) ); + " %d-byte payload with sum %s (for %s)",
msgCount, data.length, Utils.getMD5SumFor( data ),
this );
// If we get this far, we're going to assume the send has localElems = mElems;
// succeeded. Now we need to get out of the synchronized
// block because handling all the reads on the socket can
// take a long time and will block lots of stuff
localCmds = mCmds;
localGameIDs = mGameIDs;
// Now read responses
// int nCmds = mCmds.size();
// On remote, socket is being passed through
// onHandleWork. Could take a f*ck of a long time. Maybe we
// don't want this any more?
// try ( KillerIn killer = new KillerIn( socket, 5 * nCmds ) ) {
// DataInputStream inStream =
// new DataInputStream( socket.getInputStream() );
} catch ( IOException ioe ) { } catch ( IOException ioe ) {
Log.e( TAG, "writeAndCheck(): ioe: %s", ioe.getMessage() ); Log.e( TAG, "writeAndCheck(): ioe: %s", ioe.getMessage() );
} }
@ -1236,29 +1254,32 @@ public class BTService extends XWJIService {
} }
} }
Log.d( TAG, "writeAndCheck(): reading replies" );
int nDone = 0; int nDone = 0;
if ( null != localCmds ) { if ( null != localElems ) {
Log.d( TAG, "writeAndCheck(): reading %d replies", localElems.size() );
try { try {
DataInputStream inStream = DataInputStream inStream =
new DataInputStream( socket.getInputStream() ); new DataInputStream( socket.getInputStream() );
for ( int ii = 0; ii < localCmds.size(); ++ii ) { for ( int ii = 0; ii < localElems.size(); ++ii ) {
BTCmd cmd = localCmds.get(ii); MsgElem elem = localElems.get(ii);
int gameID = localGameIDs.get(ii); BTCmd cmd = elem.mCmd;
BTCmd reply = BTCmd.values()[inStream.readByte()]; int gameID = elem.mGameID;
byte cmdOrd = inStream.readByte();
if ( cmdOrd >= BTCmd.values().length ) {
break; // SNAFU!!!
}
BTCmd reply = BTCmd.values()[cmdOrd];
Log.d( TAG, "writeAndCheck() %s: got response %s to cmd[%d] %s", Log.d( TAG, "writeAndCheck() %s: got response %s to cmd[%d] %s",
this, reply, ii, cmd ); this, reply, ii, cmd );
handleReply( helper, socket, inStream, cmd, gameID, reply ); handleReply( helper, socket, inStream, cmd, gameID, reply );
nDone = localCmds.size(); ++nDone;
}
if ( nDone > 0 ) {
clearAll();
} }
} catch ( IOException ioe ) { } catch ( IOException ioe ) {
Log.d( TAG, "failed reading replies: %s", ioe.getMessage() ); Log.d( TAG, "failed reading replies: %s", ioe.getMessage() );
} }
} }
unappend( nDone );
Log.d( TAG, "writeAndCheck() => %d", nDone ); Log.d( TAG, "writeAndCheck() => %d", nDone );
return nDone; return nDone;
} }
@ -1319,16 +1340,6 @@ public class BTService extends XWJIService {
} }
} }
private void clearAll() {
Log.d( TAG, "PacketAccumulator.clearAll(): wiping %d messages",
mCmds == null ? 0 : mCmds.size() );
mOP = new OutputPair();
mCmds = new ArrayList<>();
mGameIDs = new ArrayList<>();
mFailCount = 0;
mStamp = 0;
}
void addPing( int gameID ) void addPing( int gameID )
{ {
try { try {
@ -1385,22 +1396,13 @@ public class BTService extends XWJIService {
private boolean append( BTCmd cmd, int gameID, OutputPair op ) throws IOException private boolean append( BTCmd cmd, int gameID, OutputPair op ) throws IOException
{ {
boolean haveSpace; boolean haveSpace;
// Log.d( TAG, "append() IN" );
try ( DbgUtils.DeadlockWatch dw = new DbgUtils.DeadlockWatch( this ) ) { try ( DbgUtils.DeadlockWatch dw = new DbgUtils.DeadlockWatch( this ) ) {
synchronized ( this ) { synchronized ( this ) {
haveSpace = mOP.length() + op.length() + 3 < MAX_PACKET_LEN; MsgElem newElem = new MsgElem( cmd, gameID, op );
haveSpace = mLength + newElem.size() < MAX_PACKET_LEN;
if ( haveSpace ) { if ( haveSpace ) {
if ( 0 == mCmds.size() ) { mElems.add( newElem );
mStamp = System.currentTimeMillis(); mLength += newElem.size();
}
mCmds.add( cmd );
mGameIDs.add( gameID );
mOP.dos.writeByte( cmd.ordinal() );
byte[] data = op.bos.toByteArray();
mOP.dos.writeShort( data.length );
mOP.dos.write( data, 0, data.length );
mFailCount = 0; // for now, we restart timer on new data mFailCount = 0; // for now, we restart timer on new data
tellSomebody(); tellSomebody();
} }
@ -1410,6 +1412,21 @@ public class BTService extends XWJIService {
return haveSpace; return haveSpace;
} }
private void unappend( int nToRemove )
{
Assert.assertTrue( nToRemove <= mElems.size() );
try ( DbgUtils.DeadlockWatch dw = new DbgUtils.DeadlockWatch( this ) ) {
synchronized ( this ) {
for ( int ii = 0; ii < nToRemove; ++ii ) {
MsgElem elem = mElems.remove(0);
mLength -= elem.size();
}
Log.d( TAG, "unappend(): after removing %d, have %d left for size %d",
nToRemove, mElems.size(), mLength );
}
}
}
void resetBackoff() void resetBackoff()
{ {
// Log.d( TAG, "resetBackoff() IN" ); // Log.d( TAG, "resetBackoff() IN" );
@ -1486,37 +1503,45 @@ public class BTService extends XWJIService {
data.length, Utils.getMD5SumFor( data ), nMessages ); data.length, Utils.getMD5SumFor( data ), nMessages );
for ( int ii = 0; ii < nMessages; ++ii ) { for ( int ii = 0; ii < nMessages; ++ii ) {
BTCmd cmd = BTCmd.values()[dis.readByte()]; byte cmdOrd = dis.readByte();
final short oneLen = dis.readShort(); // used only to skip short oneLen = dis.readShort(); // used only to skip
int availableBefore = dis.available(); int availableBefore = dis.available();
switch ( cmd ) { if ( cmdOrd < BTCmd.values().length ) {
case PING: BTCmd cmd = BTCmd.values()[cmdOrd];
int gameID = dis.readInt(); Log.d( TAG, "dispatchAll(): reading msg %d: %s, len=%d",
processor.receivePing( gameID, socket ); ii, cmd, oneLen );
break; switch ( cmd ) {
case INVITE: case PING:
data = new byte[dis.readShort()]; int gameID = dis.readInt();
dis.readFully( data ); processor.receivePing( gameID, socket );
NetLaunchInfo nli = XwJNI.nliFromStream( data ); break;
processor.receiveInvitation( nli, socket ); case INVITE:
break; data = new byte[dis.readShort()];
case MESG_SEND: dis.readFully( data );
gameID = dis.readInt(); NetLaunchInfo nli = XwJNI.nliFromStream( data );
data = new byte[dis.readShort()]; processor.receiveInvitation( nli, socket );
dis.readFully( data ); break;
processor.receiveMessage( gameID, data, socket ); case MESG_SEND:
break; gameID = dis.readInt();
case MESG_GAMEGONE: data = new byte[dis.readShort()];
gameID = dis.readInt(); dis.readFully( data );
processor.receiveGameGone( gameID, socket ); processor.receiveMessage( gameID, data, socket );
break; break;
default: case MESG_GAMEGONE:
Log.e( TAG, "unexpected command %s; skipping %d bytes", cmd, oneLen ); gameID = dis.readInt();
processor.receiveGameGone( gameID, socket );
break;
default:
Assert.assertFalse( BuildConfig.DEBUG );
break;
}
} else {
Log.e( TAG, "unexpected command (ord: %d);"
+ " skipping %d bytes", cmdOrd, oneLen );
if ( oneLen <= dis.available() ) { if ( oneLen <= dis.available() ) {
dis.readFully( new byte[oneLen] ); dis.readFully( new byte[oneLen] );
Assert.assertFalse( BuildConfig.DEBUG ); Assert.assertFalse( BuildConfig.DEBUG );
} }
break;
} }
// sanity-check based on packet length // sanity-check based on packet length