[reginaldalfret]

Description

This PR fixes two compatibility issues in luigi.contrib.sge:

1. Python 3 Bytes Serialization Error in _dump():
   When self.__module__ == "__main__", pickle.dumps(self) returns bytes. The code previously attempted string replacement d.replace("(c__main__", "(c" + module_name) and wrote to a file opened with mode "w". Both operations raise TypeError in Python 3 (a bytes-like object is required, not 'str' and write() argument must be str, not bytes).
   We now explicitly specify protocol 0, perform byte replacement, and open the job file in "wb" mode.

2. AttributeError / OSError with os.fstatvfs:
   In _init_local(), os.fstatvfs(0) is called to query f_namemax. On Windows or platforms/environments where fstatvfs is not supported, this raises AttributeError: module 'os' has no attribute 'fstatvfs'.
   We now defensively catch AttributeError and OSError so _init_local() proceeds cleanly.

Testing

• Added unit test test_dump_main_module in test/contrib/sge_test.py.
• All tests in test/contrib/sge_test.py pass.

[Manishram-ai]

Thanks! The bytes handling in _dump looks right. protocol=0 matters here, since protocol 2 writes c__main__ without the ( and protocol 4 and up don't write it at all, so a short comment saying that would help the next person.

A few things:

1. The other end still reads the pickle in text mode. sge_runner.py:53 opens job-instance.pickle with "r", and pickle.load on that raises TypeError: a bytes-like object is required, not 'str', so the job still fails on the compute node. Changing it to "rb" would finish the fix. The older #2480 had that change too.

2. The new test doesn't reach the rename. task.__module__ = "__main__" only sets it on the instance, so the class still pickles as sge_test and the replace doesn't change anything. A class defined in __main__ does hit it:

   main_mod = sys.modules["__main__"]
   cls = type("MainModuleJobTask", (TestJobTask,), {"__module__": "__main__"})
   setattr(main_mod, "MainModuleJobTask", cls)
   try:
       task = cls(i="2", n_cpu=1)
       with tempfile.TemporaryDirectory() as tmp_dir, patch.object(sys, "argv", ["my_pipeline.py"]):
           task._dump(tmp_dir)
           with open(task.job_file, "rb") as f:
               content = f.read()
       self.assertNotIn(b"(c__main__", content)
       self.assertIn(b"(cmy_pipeline\nMainModuleJobTask", content)
   finally:
       delattr(main_mod, "MainModuleJobTask")

   That fails on master and passes here.

3. About fstatvfs: os.fstatvfs(0) looks at stdin's filesystem, not shared_tmp_dir, and with stdin closed it raises EBADF, so catching OSError makes sense. f_namemax is also the limit for one path component, but the code cuts the whole path to that length. os.statvfs(base_tmp_dir) and trimming only folder_name (like lsf.py trims only the task name) would fix both. That could be a follow-up too.

FYI #3469 changes _dump the same way but without protocol=0, so its rename doesn't do anything.