Merge pull request #194 from cie/alexmiller/valgrind

Fix valgrind errors

fdbcli/fdbcli.actor.cpp

@@ -2674,6 +2674,7 @@ ACTOR Future<int> cli(CLIOptions opt, LineNoise* plinenoise) {
 								for (const auto& pair : interfaces) {
 									ProfilerRequest profileRequest;
 									profileRequest.type = ProfilerRequest::Type::FLOW;
+									profileRequest.action = ProfilerRequest::Action::RUN;
 									profileRequest.duration = duration;
 									profileRequest.outputFile = tokens[4];
 									all_profiler_addresses.push_back(pair.first);
@@ -2691,6 +2692,7 @@ ACTOR Future<int> cli(CLIOptions opt, LineNoise* plinenoise) {
 									for (int tokenidx = 5; tokenidx < tokens.size(); tokenidx++) {
 										ProfilerRequest profileRequest;
 										profileRequest.type = ProfilerRequest::Type::FLOW;
+										profileRequest.action = ProfilerRequest::Action::RUN;
 										profileRequest.duration = duration;
 										profileRequest.outputFile = tokens[4];
 										all_profiler_addresses.push_back(tokens[tokenidx]);

fdbclient/ClientWorkerInterface.h

@@ -80,5 +80,7 @@ struct ProfilerRequest {
 		ar & reply & type & action & duration & outputFile;
 	}
 };
+BINARY_SERIALIZABLE( ProfilerRequest::Type );
+BINARY_SERIALIZABLE( ProfilerRequest::Action );
 
 #endif

fdbclient/FailureMonitorClient.actor.cpp

@@ -139,4 +139,4 @@ ACTOR Future<Void> failureMonitorClient( Reference<AsyncVar<Optional<struct Clus
 		state Future<Void> client = ci->get().present() ? failureMonitorClientLoop(monitor, ci->get().get(), fmState, trackMyStatus) : Void();
 		Void _ = wait( ci->onChange() );
 	}
-}
+}

fdbrpc/FlowTransport.actor.cpp

@@ -28,6 +28,10 @@
 #include "crc32c.h"
 #include "simulator.h"
 
+#if VALGRIND
+#include <memcheck.h>
+#endif
+
 static NetworkAddress g_currentDeliveryPeerAddress;
 
 const UID WLTOKEN_ENDPOINT_NOT_FOUND(-1, 0);
@@ -532,6 +536,9 @@ static void scanPackets( TransportData* transport, uint8_t*& unprocessed_begin,
 			}
 		}
 
+#if VALGRIND
+		VALGRIND_CHECK_MEM_IS_DEFINED(p, packetLen);
+#endif
 		ArenaReader reader( arena, StringRef(p, packetLen), AssumeVersion(peerProtocolVersion) );
 		UID token; reader >> token;
 
@@ -807,6 +814,9 @@ static PacketID sendPacket( TransportData* self, ISerializeSource const& what, c
 		BinaryWriter wr( AssumeVersion(currentProtocolVersion) );
 		what.serializeBinaryWriter(wr);
 		Standalone<StringRef> copy = wr.toStringRef();
+#if VALGRIND
+		VALGRIND_CHECK_MEM_IS_DEFINED(copy.begin(), copy.size());
+#endif
 
 		deliver( self, destination, ArenaReader(copy.arena(), copy, AssumeVersion(currentProtocolVersion)), false );
 
@@ -891,6 +901,16 @@ static PacketID sendPacket( TransportData* self, ISerializeSource const& what, c
 				self->warnAlwaysForLargePacket = false;
 		}
 
+#if VALGRIND
+		SendBuffer *checkbuf = pb;
+		while (checkbuf) {
+			int size = checkbuf->bytes_written;
+			const uint8_t* data = checkbuf->data;
+			VALGRIND_CHECK_MEM_IS_DEFINED(data, size);
+			checkbuf = checkbuf -> next;
+		}
+#endif
+
 		peer->send(pb, rp, firstUnsent);
 
 		return (PacketID)rp;

fdbrpc/FlowTransport.h

@@ -48,6 +48,7 @@ public:
 	}
 };
 #pragma pack(pop)
+BINARY_SERIALIZABLE( Endpoint );
 
 
 class NetworkMessageReceiver {

fdbserver/workloads/CpuProfiler.actor.cpp

@@ -77,7 +77,9 @@ struct CpuProfilerWorkload : TestWorkload
 			for(i = 0; i < self->profilingWorkers.size(); i++)
 			{
 				ProfilerRequest req;
+				req.type = ProfilerRequest::Type::FLOW;
 				req.action = enabled ? ProfilerRequest::Action::ENABLE : ProfilerRequest::Action::DISABLE;
+				req.duration = 0; //unused
 
 				//The profiler output name will be the ip.port.prof
 				req.outputFile = StringRef(toIPString(self->profilingWorkers[i].address().ip) + "." + format("%d", self->profilingWorkers[i].address().port) + ".prof");
@@ -98,7 +100,9 @@ struct CpuProfilerWorkload : TestWorkload
 
 		//Enable (or disable) the profiler on the current tester
 		ProfilerRequest req;
+		req.type = ProfilerRequest::Type::FLOW;
 		req.action = enabled ? ProfilerRequest::Action::ENABLE : ProfilerRequest::Action::DISABLE;
+		req.duration = 0; //unused
 		req.outputFile = StringRef(toIPString(g_network->getLocalAddress().ip) + "." + format("%d", g_network->getLocalAddress().port) + ".prof");
 
 		updateCpuProfiler(req);

flow/serialize.h

@@ -28,6 +28,27 @@
 #include "Arena.h"
 #include <algorithm>
 
+// Though similar, is_binary_serializable cannot be replaced by std::is_pod, as doing so would prefer
+// memcpy over a defined serialize() method on a POD struct.  As not all of our structs are packed,
+// this would both inflate message sizes by transmitting padding, and mean that we're transmitting
+// undefined bytes over the wire.
+// A more intelligent SFINAE that does "binarySerialize if POD and no serialize() is defined" could
+// replace the usage of is_binary_serializable.
+template <class T>
+struct is_binary_serializable { enum { value = 0 }; };
+
+#define BINARY_SERIALIZABLE( T ) template<> struct is_binary_serializable<T> { enum { value = 1 }; };
+
+BINARY_SERIALIZABLE( uint8_t );
+BINARY_SERIALIZABLE( int16_t );
+BINARY_SERIALIZABLE( uint16_t );
+BINARY_SERIALIZABLE( int32_t );
+BINARY_SERIALIZABLE( uint32_t );
+BINARY_SERIALIZABLE( int64_t );
+BINARY_SERIALIZABLE( uint64_t );
+BINARY_SERIALIZABLE( bool );
+BINARY_SERIALIZABLE( double );
+
 template <class Archive, class Item>
 inline typename Archive::WRITER& operator << (Archive& ar, const Item& item ) {
 	save(ar, const_cast<Item&>(item));
@@ -88,7 +109,7 @@ inline void save( Archive& ar, const std::string& value ) {
 }
 
 template <class Archive, class T>
-class Serializer< Archive, T, typename std::enable_if< std::is_pod<T>::value >::type> {
+class Serializer< Archive, T, typename std::enable_if< is_binary_serializable<T>::value >::type> {
 public:
 	static void serialize( Archive& ar, T& t ) {
 		ar.serializeBinaryItem(t);