[SPARK-3993] [PySpark] fix bug while reuse worker after take()

After take(), maybe there are some garbage left in the socket, then next task assigned to this worker will hang because of corrupted data.

We should make sure the socket is clean before reuse it, write END_OF_STREAM at the end, and check it after read out all result from python.
This seems to assume that `infile` will point to the beginning of a (length, [optional data]) pair rather than at some random data in the middle of a data block.  I believe that this is actually correct given our current code, but I'm worried that the reason _why_ it's correct is subtle / brittle.

Our current serializers seem to maintain this invariant that once control flow has exited the serializer then `infile` will point to a valid length.  For example, if I call `deserializer.load_stream(infile)` then call `next()` on this iterator, then once my `next()` call returns `infile` will point to a valid length.  This invariant holds for our batched pickle serializer because we first deserialize an entire block (e.g. read a whole (length, data) pair) then yield its items one-by-one.  I think it also holds for all of the other serializers, but it could potentially be broken by a streaming deserializer which doesn't need to read the entire data block in order to yield the first item.

Basically, my concern is that there's a lot of implicit / subtle assumptions here that we should document if we're going to take this approach.